From 94c5ddeaba47908057b162b0a63dd225e79ab917 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 27 Aug 2026 14:50:34 +0000 Subject: [PATCH] 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. --- demo/demo.py | 8 +-- docs/database.md | 14 ++-- kanta/__init__.py | 17 ++++- kanta/__main__.py | 42 ++++-------- kanta/callbacks.py | 117 ++++++++++++++++++++++---------- kanta/kanta.py | 2 +- kanta/kantaimpl.py | 27 ++++---- kanta/migrations.py | 32 ++++++--- kanta/replaylog.py | 26 ++----- tests/test_callbacks.py | 102 +++++++++++++++++++++++++++- tests/test_cli.py | 7 +- tests/test_kanta_integration.py | 35 ++++++++++ tests/test_migrations.py | 19 ++++++ 13 files changed, 320 insertions(+), 128 deletions(-) diff --git a/demo/demo.py b/demo/demo.py index 629d71b..f6a68a1 100644 --- a/demo/demo.py +++ b/demo/demo.py @@ -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: diff --git a/docs/database.md b/docs/database.md index 3629e40..0ae69a4 100644 --- a/docs/database.md +++ b/docs/database.md @@ -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: diff --git a/kanta/__init__.py b/kanta/__init__.py index 84a0ea7..c8e6419 100644 --- a/kanta/__init__.py +++ b/kanta/__init__.py @@ -1,4 +1,17 @@ +from .callbacks import DictPrev, DictState, LogFmt +from .exceptions import DatabaseError from .kanta import Kanta -from .logging import configure_logging +from .logging import LogEvent, configure_logging +from .migrations import MigrationReport -__all__ = ["Kanta", "configure_logging"] +__all__ = [ + "Kanta", + "DatabaseError", + "configure_logging", + # Callback argument types + "DictPrev", + "DictState", + "LogEvent", + "LogFmt", + "MigrationReport", +] diff --git a/kanta/__main__.py b/kanta/__main__.py index 8a2e501..202a7de 100644 --- a/kanta/__main__.py +++ b/kanta/__main__.py @@ -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 diff --git a/kanta/callbacks.py b/kanta/callbacks.py index 034e997..c12a5be 100644 --- a/kanta/callbacks.py +++ b/kanta/callbacks.py @@ -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) diff --git a/kanta/kanta.py b/kanta/kanta.py index 6eca545..7e01d91 100644 --- a/kanta/kanta.py +++ b/kanta/kanta.py @@ -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. diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index bef0ebf..38b4d6e 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -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 diff --git a/kanta/migrations.py b/kanta/migrations.py index 8a2ec1b..e969990 100644 --- a/kanta/migrations.py +++ b/kanta/migrations.py @@ -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 + ) diff --git a/kanta/replaylog.py b/kanta/replaylog.py index f311307..a31c46f 100644 --- a/kanta/replaylog.py +++ b/kanta/replaylog.py @@ -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) diff --git a/tests/test_callbacks.py b/tests/test_callbacks.py index 32b0d2a..2fd4f78 100644 --- a/tests/test_callbacks.py +++ b/tests/test_callbacks.py @@ -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 diff --git a/tests/test_cli.py b/tests/test_cli.py index b28e3d6..0399464 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -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)]) diff --git a/tests/test_kanta_integration.py b/tests/test_kanta_integration.py index a59a13b..7971575 100644 --- a/tests/test_kanta_integration.py +++ b/tests/test_kanta_integration.py @@ -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" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 6517805..8cd8452 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -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