Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
94c5ddeaba | ||
|
|
4ef74e027f |
+3
-5
@@ -6,9 +6,7 @@ from pathlib import Path
|
||||
|
||||
import msgspec
|
||||
|
||||
from kanta import Kanta
|
||||
from kanta.callbacks import DictPre
|
||||
from kanta.logging import configure_logging
|
||||
from kanta import Kanta, configure_logging
|
||||
|
||||
|
||||
filename = Path(__file__).with_name("demo.kantadb")
|
||||
@@ -47,11 +45,11 @@ kanta_v1 = Kanta(filename, Data(), migrations=sys.modules[__name__])
|
||||
|
||||
|
||||
@kanta_v1.logfmt
|
||||
def resolve_user(value: str, path: str, previous: DictPre) -> str | None:
|
||||
def resolve_user(value: str, path: str, state: dict) -> str | None:
|
||||
"""Resolve user ids to names from the database state itself."""
|
||||
if path != "$user" and not path.startswith("users."):
|
||||
return None
|
||||
return previous.get("users", {}).get(value, {}).get("name")
|
||||
return state.get("users", {}).get(value, {}).get("name")
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
|
||||
+9
-5
@@ -186,9 +186,13 @@ when they have a default value.
|
||||
transaction actor, replacing the old `user_display` parameter.
|
||||
- The callback returns `str | None`: a string replaces the default rendering,
|
||||
while `None` means "fall through to the next formatter".
|
||||
- State dicts can be injected via `DictPre` (`Annotated[dict, "pre"]`)
|
||||
and `DictPost` (`Annotated[dict, "post"]`); the `Kanta` instance can also be
|
||||
injected.
|
||||
- State dicts are injected by parameter name or annotation tag, which share
|
||||
the same vocabulary: `prev` receives the previous state dict and `state`
|
||||
the current one. Matching by name ignores the annotation entirely. The
|
||||
`DictPrev`/`DictState` aliases (`Annotated[dict, "prev"]` /
|
||||
`Annotated[dict, "state"]`) work under any parameter name, and a tag takes
|
||||
precedence over the name. `DictPre` and `DictPost` are kept as aliases of
|
||||
`DictPrev` and `DictState`. The `Kanta` instance can also be injected.
|
||||
- Alternatively, a logfmt callback can be a class inheriting from `LogFmt`; the
|
||||
framework instantiates it with the state dicts and calls its
|
||||
`resolve(value, path) -> str | None` method.
|
||||
@@ -201,8 +205,8 @@ values at that exact path:
|
||||
|
||||
```python
|
||||
@kanta.logfmt(path="$user")
|
||||
def resolve_user(value: str, current: DictPost) -> str | None:
|
||||
return current.get("users", {}).get(value, {}).get("name")
|
||||
def resolve_user(value: str, state: dict) -> str | None:
|
||||
return state.get("users", {}).get(value, {}).get("name")
|
||||
|
||||
@kanta.logfmt(path="users.uuid-1")
|
||||
def resolve_user_key(value: str) -> str | None:
|
||||
|
||||
@@ -1,5 +1,17 @@
|
||||
from .callbacks import DictPrev, DictState, LogFmt
|
||||
from .exceptions import DatabaseError
|
||||
from .kanta import Kanta
|
||||
from .logging import LogEvent, configure_logging
|
||||
from .migrations import MigrationReport
|
||||
|
||||
__all__ = [
|
||||
"Kanta",
|
||||
"DatabaseError",
|
||||
"configure_logging",
|
||||
# Callback argument types
|
||||
"DictPrev",
|
||||
"DictState",
|
||||
"LogEvent",
|
||||
"LogFmt",
|
||||
"MigrationReport",
|
||||
]
|
||||
|
||||
+13
-29
@@ -129,9 +129,7 @@ def _import_kanta_object(path: str) -> Any:
|
||||
try:
|
||||
return getattr(module, "kanta")
|
||||
except AttributeError as exc:
|
||||
raise ImportError(
|
||||
f"no 'kanta' object found in {path!r}"
|
||||
) from exc
|
||||
raise ImportError(f"no 'kanta' object found in {path!r}") from exc
|
||||
try:
|
||||
spec = importlib.util.find_spec(path)
|
||||
except ImportError:
|
||||
@@ -141,9 +139,7 @@ def _import_kanta_object(path: str) -> Any:
|
||||
try:
|
||||
return getattr(module, "kanta")
|
||||
except AttributeError as exc:
|
||||
raise ImportError(
|
||||
f"no 'kanta' object found in module {path!r}"
|
||||
) from exc
|
||||
raise ImportError(f"no 'kanta' object found in module {path!r}") from exc
|
||||
return _import_dotted(path)
|
||||
|
||||
|
||||
@@ -344,16 +340,14 @@ async def _log_migration(
|
||||
try:
|
||||
await registry.invoke(
|
||||
"logmigr",
|
||||
InjectionContext(kanta=kanta, migration_result=result),
|
||||
InjectionContext(kanta=kanta, report=result),
|
||||
)
|
||||
except Exception:
|
||||
_logger.exception("logmigr callback failed")
|
||||
return
|
||||
if quiet:
|
||||
return
|
||||
descriptions = [
|
||||
f"{m.name} ({m.description})" for m in result.migrations if m.changed
|
||||
]
|
||||
descriptions = [f"{m.name} ({m.description})" for m in result.applied if m.changed]
|
||||
emit_event(
|
||||
LogEvent(
|
||||
kind="migrated",
|
||||
@@ -369,9 +363,7 @@ async def _log_migration(
|
||||
)
|
||||
|
||||
|
||||
def _get_kanta(
|
||||
args: argparse.Namespace, filename: Path
|
||||
) -> tuple[Kanta[Any], bool]:
|
||||
def _get_kanta(args: argparse.Namespace, filename: Path) -> tuple[Kanta[Any], bool]:
|
||||
"""Return the Kanta instance to work with, and whether the CLI owns it.
|
||||
|
||||
With ``-k`` the existing object is used as-is (and never closed by us);
|
||||
@@ -391,9 +383,7 @@ def _get_kanta(
|
||||
return Kanta(filename, {}, type=dict, migrations=args.migrations), True
|
||||
except Exception as exc:
|
||||
if args.migrations:
|
||||
raise _CliError(
|
||||
f"Migration error: {exc}", EXIT_MIGRATION_ERROR
|
||||
) from exc
|
||||
raise _CliError(f"Migration error: {exc}", EXIT_MIGRATION_ERROR) from exc
|
||||
raise _CliError(f"Failed to initialize database: {exc}") from exc
|
||||
|
||||
|
||||
@@ -473,9 +463,7 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
state = {}
|
||||
version = 0
|
||||
printed = False
|
||||
for event, previous, current in replay_events(
|
||||
events, selection.end_line
|
||||
):
|
||||
for event, previous, current in replay_events(events, selection.end_line):
|
||||
state = current
|
||||
version = event.version
|
||||
if event.line_number < selection.start_line or args.quiet:
|
||||
@@ -506,7 +494,9 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
f"Migration error: {exc}", EXIT_MIGRATION_ERROR
|
||||
) from exc
|
||||
if version != previous_version:
|
||||
await _log_migration(kanta, filename, result, previous_version, args.quiet)
|
||||
await _log_migration(
|
||||
kanta, filename, result, previous_version, args.quiet
|
||||
)
|
||||
|
||||
output_state: dict[str, Any]
|
||||
if data_type is not None:
|
||||
@@ -529,9 +519,7 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
# the object's own serializer, and its migrations were applied
|
||||
# to the state; no need to re-open through a new instance.
|
||||
print(f"{data}", file=sys.stderr)
|
||||
output_state = struct_to_dict(
|
||||
data, serializer=kanta._impl.serializer
|
||||
)
|
||||
output_state = struct_to_dict(data, serializer=kanta._impl.serializer)
|
||||
else:
|
||||
kanta_typed = Kanta(
|
||||
filename, data, type=data_type, migrations=args.migrations
|
||||
@@ -544,9 +532,7 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
f"Validation error: {exc}", EXIT_VALIDATION_ERROR
|
||||
) from exc
|
||||
except DataIntegrityError as exc:
|
||||
raise _CliError(
|
||||
f"Parse error: {exc}", EXIT_PARSE_ERROR
|
||||
) from exc
|
||||
raise _CliError(f"Parse error: {exc}", EXIT_PARSE_ERROR) from exc
|
||||
except DatabaseError as exc:
|
||||
if not args.migrations or exc.cause_type == "ReplayError":
|
||||
raise _CliError(
|
||||
@@ -560,9 +546,7 @@ async def _run(args: argparse.Namespace) -> int:
|
||||
raise _CliError(
|
||||
f"Migration error: {exc}", EXIT_MIGRATION_ERROR
|
||||
) from exc
|
||||
raise _CliError(
|
||||
f"Failed to open {filename}: {exc}"
|
||||
) from exc
|
||||
raise _CliError(f"Failed to open {filename}: {exc}") from exc
|
||||
output_state = kanta_typed._impl.statedict
|
||||
else:
|
||||
output_state = state
|
||||
|
||||
+80
-37
@@ -1,8 +1,8 @@
|
||||
"""Unified decorator-based callback registry for Kanta.
|
||||
|
||||
Callbacks are registered once and invoked with arguments filled by their
|
||||
annotation types. Unknown arguments are only permitted when they have a
|
||||
default value.
|
||||
Callbacks are registered once and invoked with arguments filled from their
|
||||
parameter names (state dicts: ``prev`` / ``state``) and annotation types.
|
||||
Unknown arguments are only permitted when they have a default value.
|
||||
|
||||
Log formatters are a special case: they are called per value being rendered
|
||||
and receive the value plus an optional ``path`` string. They return
|
||||
@@ -23,10 +23,36 @@ from dataclasses import dataclass
|
||||
from typing import Annotated, Any, Union, get_args, get_origin
|
||||
|
||||
from kanta.exceptions import DatabaseError
|
||||
from kanta.migrations import MigrationResult
|
||||
from kanta.migrations import MigrationReport
|
||||
|
||||
DictPrev = DictPre = Annotated[dict, "prev"]
|
||||
DictState = DictPost = Annotated[dict, "state"]
|
||||
|
||||
# State-dict injection keys, shared by parameter names and annotation tags:
|
||||
# a callback parameter named *or* tagged ``prev``/``state`` receives the
|
||||
# previous or current state dict respectively. Matching by name does not
|
||||
# check the annotation; an explicit tag takes precedence over the name.
|
||||
_STATE_KINDS = {"prev": "previous_state", "state": "current_state"}
|
||||
|
||||
|
||||
def _state_key(text: Any) -> str | None:
|
||||
"""Return the state kind for a parameter name or annotation tag."""
|
||||
return text if text in _STATE_KINDS else None
|
||||
|
||||
|
||||
def _state_tag(ann: Any) -> str | None:
|
||||
"""Return the state tag of an ``Annotated[dict, ...]`` annotation, if any."""
|
||||
if get_origin(ann) is not Annotated:
|
||||
return None
|
||||
args = get_args(ann)
|
||||
if not args or args[0] is not dict:
|
||||
return None
|
||||
for meta in args[1:]:
|
||||
key = _state_key(meta)
|
||||
if key is not None:
|
||||
return key
|
||||
return None
|
||||
|
||||
DictPre = Annotated[dict, "pre"]
|
||||
DictPost = Annotated[dict, "post"]
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -35,16 +61,18 @@ class LogFmt:
|
||||
"""Base class for stateful logfmt callbacks.
|
||||
|
||||
Subclasses only need to override :meth:`resolve`. The framework injects
|
||||
``previous_state`` and ``current_state`` through ``__init__``.
|
||||
the previous and current state dicts through ``__init__`` and exposes them
|
||||
as ``previous_state`` and ``state``.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
previous: DictPre | None = None,
|
||||
current: DictPost | None = None,
|
||||
prev: dict | None = None,
|
||||
state: dict | None = None,
|
||||
) -> None:
|
||||
self.previous_state = previous
|
||||
self.current_state = current
|
||||
self.previous_state = prev
|
||||
self.state = state
|
||||
self.current_state = state # deprecated alias for ``state``
|
||||
|
||||
def __call__(self, value: Any, path: str) -> str | None:
|
||||
return self.resolve(value, path)
|
||||
@@ -67,7 +95,7 @@ class InjectionContext:
|
||||
error: DatabaseError | None = None
|
||||
previous_state: dict | None = None
|
||||
current_state: dict | None = None
|
||||
migration_result: MigrationResult | None = None
|
||||
report: MigrationReport | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -261,6 +289,9 @@ class CallbackRegistry:
|
||||
)
|
||||
|
||||
if param.annotation is inspect.Parameter.empty:
|
||||
if _state_key(name) is not None:
|
||||
params.append((name, dict))
|
||||
continue
|
||||
if param.default is inspect.Parameter.empty:
|
||||
raise TypeError(
|
||||
f"{kind} callback {callback.__name__} has parameter "
|
||||
@@ -269,6 +300,11 @@ class CallbackRegistry:
|
||||
continue
|
||||
|
||||
ann = self._resolve_raw_annotation(param.annotation, callback)
|
||||
if _state_key(name) is not None:
|
||||
# The name alone selects state injection; an explicit tag
|
||||
# still overrides it. The annotation is not checked.
|
||||
params.append((name, ann))
|
||||
continue
|
||||
if not self._is_allowed(kind, ann):
|
||||
if param.default is inspect.Parameter.empty:
|
||||
raise TypeError(
|
||||
@@ -329,6 +365,9 @@ class CallbackRegistry:
|
||||
f"*args or **kwargs"
|
||||
)
|
||||
if param.annotation is inspect.Parameter.empty:
|
||||
if _state_key(name) is not None:
|
||||
inject_params.append((name, dict))
|
||||
continue
|
||||
if param.default is inspect.Parameter.empty:
|
||||
raise TypeError(
|
||||
f"logfmt callback {callback.__name__} has parameter "
|
||||
@@ -337,6 +376,11 @@ class CallbackRegistry:
|
||||
continue
|
||||
|
||||
ann = self._resolve_raw_annotation(param.annotation, callback)
|
||||
if _state_key(name) is not None:
|
||||
# The name alone selects state injection; an explicit tag
|
||||
# still overrides it. The annotation is not checked.
|
||||
inject_params.append((name, ann))
|
||||
continue
|
||||
if name == "path" and self._unwrap_optional(ann) is str:
|
||||
has_path = True
|
||||
continue
|
||||
@@ -392,6 +436,9 @@ class CallbackRegistry:
|
||||
f"*args or **kwargs"
|
||||
)
|
||||
if param.annotation is inspect.Parameter.empty:
|
||||
if _state_key(name) is not None:
|
||||
inject_params.append((name, dict))
|
||||
continue
|
||||
if param.default is inspect.Parameter.empty:
|
||||
raise TypeError(
|
||||
f"logfmt class {cls.__name__}.__init__ has parameter "
|
||||
@@ -400,6 +447,11 @@ class CallbackRegistry:
|
||||
continue
|
||||
|
||||
ann = self._resolve_raw_annotation(param.annotation, cls.__init__)
|
||||
if _state_key(name) is not None:
|
||||
# The name alone selects state injection; an explicit tag
|
||||
# still overrides it. The annotation is not checked.
|
||||
inject_params.append((name, ann))
|
||||
continue
|
||||
if self._is_allowed("logfmt", ann):
|
||||
inject_params.append((name, ann))
|
||||
continue
|
||||
@@ -464,7 +516,7 @@ class CallbackRegistry:
|
||||
) -> dict[str, Any]:
|
||||
kwargs: dict[str, Any] = {}
|
||||
for name, ann in params:
|
||||
value = self._resolve_annotation(ann, ctx)
|
||||
value = self._resolve_annotation(name, ann, ctx)
|
||||
if value is _UNRESOLVED:
|
||||
raise RuntimeError(f"no value available for annotation {ann!r}")
|
||||
kwargs[name] = value
|
||||
@@ -472,13 +524,11 @@ class CallbackRegistry:
|
||||
|
||||
def _is_allowed(self, kind: str, ann: Any) -> bool:
|
||||
bare = self._unwrap_optional(ann)
|
||||
if self._matches_state_annotation(bare, "pre"):
|
||||
return kind == "logfmt"
|
||||
if self._matches_state_annotation(bare, "post"):
|
||||
if _state_tag(bare) is not None:
|
||||
return kind == "logfmt"
|
||||
if bare is DatabaseError:
|
||||
return kind == "fatal_error"
|
||||
if bare is MigrationResult:
|
||||
if bare is MigrationReport:
|
||||
return kind == "logmigr"
|
||||
if self._data_type is not None and bare is self._data_type:
|
||||
return kind == "bootstrap"
|
||||
@@ -502,22 +552,25 @@ class CallbackRegistry:
|
||||
if kind == "fatal_error":
|
||||
parts.append("DatabaseError")
|
||||
if kind == "logmigr":
|
||||
parts.append("MigrationResult")
|
||||
parts.append("MigrationReport")
|
||||
if kind == "logfmt":
|
||||
parts.append("Annotated[dict, 'pre']")
|
||||
parts.append("Annotated[dict, 'post']")
|
||||
parts.append("prev: dict")
|
||||
parts.append("state: dict")
|
||||
return ", ".join(parts) if parts else "none"
|
||||
|
||||
def _resolve_annotation(self, ann: Any, ctx: InjectionContext) -> Any:
|
||||
def _resolve_annotation(self, name: str, ann: Any, ctx: InjectionContext) -> Any:
|
||||
bare = self._unwrap_optional(ann)
|
||||
if self._matches_state_annotation(bare, "pre"):
|
||||
return ctx.previous_state
|
||||
if self._matches_state_annotation(bare, "post"):
|
||||
return ctx.current_state
|
||||
# An explicit tag takes precedence over the parameter name.
|
||||
tag = _state_tag(bare)
|
||||
if tag is not None:
|
||||
return getattr(ctx, _STATE_KINDS[tag])
|
||||
key = _state_key(name)
|
||||
if key is not None:
|
||||
return getattr(ctx, _STATE_KINDS[key])
|
||||
if bare is DatabaseError:
|
||||
return ctx.error
|
||||
if bare is MigrationResult:
|
||||
return ctx.migration_result
|
||||
if bare is MigrationReport:
|
||||
return ctx.report
|
||||
if self._data_type is not None and bare is self._data_type:
|
||||
return ctx.data
|
||||
if self._kanta_class is not None and bare is self._kanta_class:
|
||||
@@ -539,16 +592,6 @@ class CallbackRegistry:
|
||||
) from exc
|
||||
return raw_ann
|
||||
|
||||
@staticmethod
|
||||
def _matches_state_annotation(ann: Any, marker: str) -> bool:
|
||||
origin = get_origin(ann)
|
||||
if origin is not Annotated:
|
||||
return False
|
||||
args = get_args(ann)
|
||||
if not args:
|
||||
return False
|
||||
return args[0] is dict and marker in args[1:]
|
||||
|
||||
@staticmethod
|
||||
def _unwrap_optional(ann: Any) -> Any:
|
||||
origin = get_origin(ann)
|
||||
|
||||
+1
-1
@@ -281,7 +281,7 @@ class Kanta(Generic[T]):
|
||||
"""Register a migration logging callback.
|
||||
|
||||
Can be used as ``@kanta.logmigr``.
|
||||
The callback receives a :class:`kanta.migrations.MigrationResult` and
|
||||
The callback receives a :class:`kanta.migrations.MigrationReport` and
|
||||
may be sync or async. If registered, it replaces the default migration
|
||||
logger output; the application is responsible for emitting any log
|
||||
messages.
|
||||
|
||||
+12
-15
@@ -19,7 +19,7 @@ from kanta.logging import (
|
||||
emit_event,
|
||||
migration_logger,
|
||||
)
|
||||
from kanta.migrations import MigrationResult, Migrations
|
||||
from kanta.migrations import MigrationReport, Migrations
|
||||
from kanta.persistence import PersistenceMixin
|
||||
from kanta.serialization import restore_data_in_place, struct_to_dict
|
||||
from kanta.serialization.base import replay
|
||||
@@ -97,19 +97,18 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
||||
|
||||
async def _handle_migration_log(
|
||||
self,
|
||||
migration_result: MigrationResult,
|
||||
previous_version: int,
|
||||
report: MigrationReport,
|
||||
log: bool | logging.Logger,
|
||||
) -> None:
|
||||
"""Route migration logging to callback or default logger."""
|
||||
assert isinstance(migration_result, MigrationResult)
|
||||
assert isinstance(report, MigrationReport)
|
||||
|
||||
if self.callback_registry.has("logmigr"):
|
||||
await self.callback_registry.invoke(
|
||||
"logmigr",
|
||||
InjectionContext(
|
||||
kanta=self._kanta,
|
||||
migration_result=migration_result,
|
||||
report=report,
|
||||
),
|
||||
on_error=_log_callback_error,
|
||||
)
|
||||
@@ -120,7 +119,7 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
||||
|
||||
migration_log = log if isinstance(log, logging.Logger) else migration_logger
|
||||
|
||||
changed = [m for m in migration_result.migrations if m.changed]
|
||||
changed = [m for m in report.applied if m.changed]
|
||||
if not changed:
|
||||
return
|
||||
|
||||
@@ -131,8 +130,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
||||
logger=migration_log,
|
||||
kanta=self._kanta,
|
||||
filename=str(self.filename),
|
||||
from_version=previous_version,
|
||||
to_version=migration_result.version,
|
||||
from_version=report.original,
|
||||
to_version=report.version,
|
||||
migrations=descriptions,
|
||||
),
|
||||
self.callback_registry.logemit_handlers,
|
||||
@@ -211,15 +210,15 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
||||
cause_type=type(e).__name__,
|
||||
) from e
|
||||
|
||||
migration_result = None
|
||||
migration_report = None
|
||||
state_before_migrations = None
|
||||
previous_version = rr.version
|
||||
if self.migrations is not None:
|
||||
state_before_migrations = copy.deepcopy(rr.state)
|
||||
migration_result = self.migrations.apply(
|
||||
migration_report = self.migrations.apply(
|
||||
rr.state, rr.version, self._kanta
|
||||
)
|
||||
rr.version = migration_result.version
|
||||
rr.version = migration_report.version
|
||||
|
||||
migrations_ran = rr.version != previous_version
|
||||
|
||||
@@ -267,10 +266,8 @@ class KantaImpl(PersistenceMixin, Generic[T]):
|
||||
)
|
||||
record = self.queue_change(action, normalized, mtime=False)
|
||||
# The migration summary introduces the diff, so log it first.
|
||||
if migrations_ran and migration_result is not None:
|
||||
await self._handle_migration_log(
|
||||
migration_result, previous_version, log
|
||||
)
|
||||
if migrations_ran and migration_report is not None:
|
||||
await self._handle_migration_log(migration_report, log)
|
||||
if (
|
||||
record is not None
|
||||
and log is not False
|
||||
|
||||
+22
-10
@@ -34,11 +34,20 @@ class MigrationInfo:
|
||||
|
||||
|
||||
@dataclass
|
||||
class MigrationResult:
|
||||
"""Result of applying migrations."""
|
||||
class MigrationReport:
|
||||
"""Report of applying migrations."""
|
||||
|
||||
version: int
|
||||
migrations: list[MigrationInfo]
|
||||
original: int
|
||||
applied: list[MigrationInfo]
|
||||
|
||||
@property
|
||||
def migrations(self) -> list[MigrationInfo]:
|
||||
"""Deprecated alias for :attr:`applied`."""
|
||||
return self.applied
|
||||
|
||||
|
||||
MigrationResult = MigrationReport # deprecated alias for MigrationReport
|
||||
|
||||
|
||||
class Migrations:
|
||||
@@ -57,13 +66,13 @@ class Migrations:
|
||||
def migrate_v2(d: dict) -> None:
|
||||
d.setdefault("version", 2)
|
||||
|
||||
result = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
new_version = result.version
|
||||
report = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
new_version = report.version
|
||||
|
||||
Or load from a module::
|
||||
|
||||
migrations = Migrations.from_module("myapp.migrations")
|
||||
result = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
report = migrations.apply(state, current_version=0, kanta=kanta)
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
@@ -137,7 +146,7 @@ class Migrations:
|
||||
data_dict: dict[str, Any],
|
||||
current_version: int,
|
||||
kanta: Any,
|
||||
) -> MigrationResult:
|
||||
) -> MigrationReport:
|
||||
"""Apply pending migrations to *data_dict* in place.
|
||||
|
||||
Missing intermediate migration steps are silently skipped.
|
||||
@@ -146,8 +155,8 @@ class Migrations:
|
||||
DatabaseError: If the database version is newer than the highest
|
||||
supported version or older than the minimum supported version.
|
||||
|
||||
Returns a :class:`MigrationResult` describing the new version and every
|
||||
migration that ran.
|
||||
Returns a :class:`MigrationReport` describing the original and new
|
||||
versions and every migration that ran.
|
||||
"""
|
||||
if current_version > self.dbver:
|
||||
raise DatabaseError(
|
||||
@@ -161,6 +170,7 @@ class Migrations:
|
||||
)
|
||||
|
||||
migrations: list[MigrationInfo] = []
|
||||
original = current_version
|
||||
for version in sorted(self._migrations.keys()):
|
||||
if version <= current_version:
|
||||
continue
|
||||
@@ -181,4 +191,6 @@ class Migrations:
|
||||
before=before,
|
||||
)
|
||||
)
|
||||
return MigrationResult(version=current_version, migrations=migrations)
|
||||
return MigrationReport(
|
||||
version=current_version, original=original, applied=migrations
|
||||
)
|
||||
|
||||
+7
-19
@@ -113,15 +113,11 @@ def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
|
||||
record_type="snapshot",
|
||||
)
|
||||
state = snap.state
|
||||
events.append(
|
||||
SnapshotEvent(line_number, byte_pos, record_index, snap)
|
||||
)
|
||||
events.append(SnapshotEvent(line_number, byte_pos, record_index, snap))
|
||||
else:
|
||||
record = impl.serializer.decode(payload, type=ChangeRecord)
|
||||
state = patch_state(state, record.diff)
|
||||
events.append(
|
||||
ChangeEvent(line_number, byte_pos, record_index, record)
|
||||
)
|
||||
events.append(ChangeEvent(line_number, byte_pos, record_index, record))
|
||||
change_count += 1
|
||||
except msgspec.DecodeError as exc:
|
||||
raise ReplayError(
|
||||
@@ -294,17 +290,13 @@ def _bound_to_line(
|
||||
raise ValueError(f"unknown range unit: {unit}")
|
||||
|
||||
|
||||
def _resolve_range(
|
||||
range_str: str, events: list[Event], total: int
|
||||
) -> tuple[int, int]:
|
||||
def _resolve_range(range_str: str, events: list[Event], total: int) -> tuple[int, int]:
|
||||
"""Parse a range string into a [start_line, end_line) line range."""
|
||||
sep = ".." if ".." in range_str else ":"
|
||||
start_str, end_str = range_str.split(sep, 1)
|
||||
start_unit, start_val = _parse_bound(start_str)
|
||||
end_unit, end_val = _parse_bound(end_str)
|
||||
start_line = _bound_to_line(
|
||||
start_unit, start_val, events, total, is_start=True
|
||||
)
|
||||
start_line = _bound_to_line(start_unit, start_val, events, total, is_start=True)
|
||||
end_line = _bound_to_line(end_unit, end_val, events, total, is_start=False)
|
||||
# ``..`` makes the end bound inclusive.
|
||||
if sep == ".." and end_val is not None:
|
||||
@@ -341,9 +333,7 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
|
||||
n_snapshots = sum(isinstance(e, SnapshotEvent) for e in events)
|
||||
count = _plural(n_snapshots, "snapshot")
|
||||
if not 0 <= idx < len(lines):
|
||||
raise RangeNotFoundError(
|
||||
f"Snapshot {spec!r} not found in file ({count})"
|
||||
)
|
||||
raise RangeNotFoundError(f"Snapshot {spec!r} not found in file ({count})")
|
||||
event = _event_at_line(events, lines[idx])
|
||||
if event is None:
|
||||
# s0 with an empty initial state (l0): not a real record, so it
|
||||
@@ -372,8 +362,7 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
|
||||
lines = _version_lines(events)
|
||||
if value not in lines:
|
||||
raise RangeNotFoundError(
|
||||
f"Version {spec!r} not found in file"
|
||||
f" ({_plural(len(lines), 'version')})"
|
||||
f"Version {spec!r} not found in file ({_plural(len(lines), 'version')})"
|
||||
)
|
||||
start_line = lines[value]
|
||||
later = [line for line in lines.values() if line > start_line]
|
||||
@@ -383,8 +372,7 @@ def select(spec: str, events: list[Event], total: int) -> Selection:
|
||||
idx = total + value if value < 0 else value
|
||||
if not 0 <= idx < total:
|
||||
raise RangeNotFoundError(
|
||||
f"Change index {spec!r} not found in file"
|
||||
f" ({_plural(total, 'change')})"
|
||||
f"Change index {spec!r} not found in file ({_plural(total, 'change')})"
|
||||
)
|
||||
end_line = lines[idx + 1] if idx + 1 < total else end_of_file(events)
|
||||
return Selection(lines[idx], end_line)
|
||||
|
||||
+101
-1
@@ -2,7 +2,7 @@ from typing import Any, Optional, Union
|
||||
|
||||
import pytest
|
||||
|
||||
from kanta import Kanta
|
||||
from kanta import DictPrev, DictState, Kanta
|
||||
from kanta.callbacks import DictPost, DictPre, LogFmt
|
||||
from kanta.exceptions import DatabaseError
|
||||
|
||||
@@ -166,6 +166,106 @@ async def test_logfmt_injects_states(tmp_path, format_config, caplog):
|
||||
assert "Alice" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logfmt_injects_states_by_name(tmp_path, format_config, caplog):
|
||||
import logging
|
||||
|
||||
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
@kanta.logfmt
|
||||
def resolve_users(value: str, prev, state: dict | None) -> str | None:
|
||||
assert prev == {}
|
||||
assert state is not None
|
||||
return state.get("users", {}).get(value, {}).get("name")
|
||||
|
||||
await kanta.open()
|
||||
|
||||
with kanta.transaction(action="create_user") as data:
|
||||
data.users["uuid-9"] = User(name="Carol")
|
||||
|
||||
await kanta.close()
|
||||
|
||||
assert "Carol" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logfmt_state_name_ignores_annotation(tmp_path, format_config, caplog):
|
||||
import logging
|
||||
|
||||
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
# Matching by name does not check the annotation.
|
||||
@kanta.logfmt
|
||||
def resolve_users(value: str, state: int) -> str | None:
|
||||
return state.get("users", {}).get(value, {}).get("name")
|
||||
|
||||
await kanta.open()
|
||||
|
||||
with kanta.transaction(action="create_user") as data:
|
||||
data.users["uuid-10"] = User(name="Dave")
|
||||
|
||||
await kanta.close()
|
||||
|
||||
assert "Dave" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logfmt_tag_takes_precedence_over_name(tmp_path, format_config, caplog):
|
||||
import logging
|
||||
|
||||
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
@kanta.logfmt
|
||||
def check_prev(value: str, anything: DictPrev) -> str | None:
|
||||
assert anything == {}
|
||||
return None
|
||||
|
||||
@kanta.logfmt
|
||||
def resolve_users(value: str, prev: DictState) -> str | None:
|
||||
# The tag wins: prev receives the current state despite its name.
|
||||
return prev.get("users", {}).get(value, {}).get("name")
|
||||
|
||||
await kanta.open()
|
||||
|
||||
with kanta.transaction(action="create_user") as data:
|
||||
data.users["uuid-11"] = User(name="Erin")
|
||||
|
||||
await kanta.close()
|
||||
|
||||
assert "Erin" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logfmt_class_state_attribute(tmp_path, format_config, caplog):
|
||||
import logging
|
||||
|
||||
caplog.set_level(logging.INFO, logger="kanta.transaction")
|
||||
path = tmp_path / "test.db"
|
||||
kanta = make_kanta(path, Data, format_config)
|
||||
|
||||
@kanta.logfmt
|
||||
class UserLogFmt(LogFmt):
|
||||
def resolve(self, value: str, path: str) -> str | None:
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
return self.state.get("users", {}).get(value, {}).get("name")
|
||||
|
||||
await kanta.open()
|
||||
|
||||
with kanta.transaction(action="create_user") as data:
|
||||
data.users["uuid-12"] = User(name="Fred")
|
||||
|
||||
await kanta.close()
|
||||
|
||||
assert "Fred" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logfmt_class_injection(tmp_path, format_config, caplog):
|
||||
import logging
|
||||
|
||||
+3
-4
@@ -86,10 +86,9 @@ def test_cli_snapshot_line_format(tmp_path, capsys):
|
||||
|
||||
snapshot = Snapshot(ts=ts, v=1, m=mtime, state={"counter": 5})
|
||||
change = ChangeRecord(ts=ts, a="inc", v=1, u="user1", diff={"counter": 6})
|
||||
data = (
|
||||
framer.frame_snapshot(serializer.encode(snapshot), record_offset=0)
|
||||
+ framer.frame_change(serializer.encode(change), record_offset=0)
|
||||
)
|
||||
data = framer.frame_snapshot(
|
||||
serializer.encode(snapshot), record_offset=0
|
||||
) + framer.frame_change(serializer.encode(change), record_offset=0)
|
||||
path.write_bytes(data)
|
||||
|
||||
code = main([str(path)])
|
||||
|
||||
@@ -721,6 +721,41 @@ async def test_logmigr_callback_replaces_default_logging(
|
||||
assert not info_messages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logmigr_callback_report(tmp_path, format_config, caplog):
|
||||
import logging
|
||||
|
||||
from kanta import MigrationReport
|
||||
|
||||
path = tmp_path / "test.db"
|
||||
seed_single_change(path, fixed_change("init", {"counter": 0}), format_config)
|
||||
|
||||
mod = type(sys)("test_migrations_report")
|
||||
|
||||
def migrate_v1(d):
|
||||
"""Bump counter."""
|
||||
d["counter"] = 2
|
||||
|
||||
mod.__dict__["migrate_v1"] = migrate_v1
|
||||
|
||||
reports = []
|
||||
|
||||
kanta = make_kanta(path, Data, format_config, migrations=mod)
|
||||
|
||||
@kanta.logmigr
|
||||
def collect(report: MigrationReport):
|
||||
reports.append(report)
|
||||
|
||||
with caplog.at_level(logging.INFO, logger="kanta.migration"):
|
||||
await kanta.open()
|
||||
await kanta.close()
|
||||
|
||||
assert len(reports) == 1
|
||||
assert reports[0].original == 0
|
||||
assert reports[0].version == 1
|
||||
assert [m.name for m in reports[0].applied] == ["migrate_v1"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transaction_log_false_suppresses_log(tmp_path, format_config, caplog):
|
||||
path = tmp_path / "test.db"
|
||||
|
||||
@@ -206,3 +206,22 @@ def test_description_defaults_to_version_when_no_docstring():
|
||||
|
||||
result = reg.apply({}, current_version=0, kanta=kanta)
|
||||
assert result.migrations[0].description == "v1"
|
||||
|
||||
|
||||
def test_report_fields():
|
||||
from kanta import MigrationReport
|
||||
|
||||
reg = Migrations()
|
||||
kanta = _DummyKanta()
|
||||
|
||||
@reg.register
|
||||
def migrate_v1(d):
|
||||
d["x"] = 1
|
||||
|
||||
report = reg.apply({"x": 0}, current_version=0, kanta=kanta)
|
||||
assert isinstance(report, MigrationReport)
|
||||
assert report.original == 0
|
||||
assert report.version == 1
|
||||
assert [m.name for m in report.applied] == ["migrate_v1"]
|
||||
# Deprecated alias still works.
|
||||
assert report.migrations is report.applied
|
||||
|
||||
Reference in New Issue
Block a user