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