From 1e9f83c8005bdbeb491d752600674bc000e5800e Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 6 Aug 2026 23:03:24 +0000 Subject: [PATCH] Add rich transaction log headers: extra metadata, logheader callback, header/diff toggles, green add paths --- kanta/callbacks.py | 74 +++++++++++++++++++++++++++++++++++++++++--- kanta/kanta.py | 39 +++++++++++++++++++++-- kanta/kantaimpl.py | 46 +++++++++++++++++++++++++++ kanta/logging.py | 59 +++++++++++++++++++++++++++++------ kanta/transaction.py | 23 ++++++++++++-- 5 files changed, 220 insertions(+), 21 deletions(-) diff --git a/kanta/callbacks.py b/kanta/callbacks.py index ef82271..0d83362 100644 --- a/kanta/callbacks.py +++ b/kanta/callbacks.py @@ -61,6 +61,9 @@ class InjectionContext: previous_state: dict | None = None current_state: dict | None = None migration_result: MigrationResult | None = None + action: str | None = None + user: str | None = None + extra: Any = None @dataclass @@ -101,6 +104,7 @@ class CallbackRegistry: "bootstrap": [], "fatal_error": [], "logmigr": [], + "logheader": [], } self._logfmt_callbacks: list[_LogFmtFunctionSpec | _LogFmtClassSpec] = [] @@ -131,6 +135,18 @@ class CallbackRegistry: if not callable(callback): raise TypeError(f"{kind} callback must be callable") + if kind == "logheader": + if inspect.iscoroutinefunction(callback): + raise TypeError("logheader callbacks must not be async") + return_ann = inspect.signature(callback).return_annotation + if return_ann is not inspect.Signature.empty: + resolved = self._resolve_raw_annotation(return_ann, callback) + if not self._is_optional_str(resolved): + raise TypeError( + f"logheader callback {callback.__name__} must return " + f"str | None, got {resolved!r}" + ) + params = self._validate_function(callback, kind) is_async = inspect.iscoroutinefunction(callback) @@ -217,6 +233,27 @@ class CallbackRegistry: return format_value + def resolve_logheader(self, ctx: InjectionContext) -> str | None: + """Invoke logheader callbacks; the first non-None result wins.""" + for reg in self._callbacks["logheader"]: + kwargs = self._build_kwargs(reg.params, ctx) + result = reg.callback(**kwargs) + if result is not None: + return result + return None + + @staticmethod + def _logheader_param_annotation(name: str, ann: Any) -> Any: + """Map logheader parameter names to their injection sentinels.""" + bare = CallbackRegistry._unwrap_optional(ann) + if name == "action" and bare is str: + return _HeaderAction + if name == "user" and bare is str: + return _HeaderUser + if name == "extra" and (bare is dict or get_origin(bare) is dict): + return _HeaderExtra + return None + def _validate_function( self, callback: Callable[..., Any], @@ -240,6 +277,11 @@ class CallbackRegistry: continue ann = self._resolve_raw_annotation(param.annotation, callback) + if kind == "logheader": + header_ann = self._logheader_param_annotation(name, ann) + if header_ann is not None: + params.append((name, header_ann)) + continue if not self._is_allowed(kind, ann): if param.default is inspect.Parameter.empty: raise TypeError( @@ -444,9 +486,9 @@ 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" + return kind in {"logfmt", "logheader"} if self._matches_state_annotation(bare, "post"): - return kind == "logfmt" + return kind in {"logfmt", "logheader"} if bare is DatabaseError: return kind == "fatal_error" if bare is MigrationResult: @@ -454,7 +496,7 @@ class CallbackRegistry: if self._data_type is not None and bare is self._data_type: return kind == "bootstrap" if self._kanta_class is not None and bare is self._kanta_class: - return kind in {"bootstrap", "fatal_error", "logfmt", "logmigr"} + return kind in {"bootstrap", "fatal_error", "logfmt", "logmigr", "logheader"} return False def _allowed_message(self, kind: str) -> str: @@ -462,19 +504,29 @@ class CallbackRegistry: if kind == "bootstrap": if self._data_type is not None: parts.append(self._data_type.__name__) - if kind in {"bootstrap", "fatal_error", "logfmt"}: + if kind in {"bootstrap", "fatal_error", "logfmt", "logmigr", "logheader"}: if self._kanta_class is not None: parts.append(self._kanta_class.__name__) if kind == "fatal_error": parts.append("DatabaseError") if kind == "logmigr": parts.append("MigrationResult") - if kind == "logfmt": + if kind == "logheader": + parts.append("action: str") + parts.append("user: str | None") + parts.append("extra: dict | None") + if kind in {"logfmt", "logheader"}: parts.append("Annotated[dict, 'pre']") parts.append("Annotated[dict, 'post']") return ", ".join(parts) if parts else "none" def _resolve_annotation(self, ann: Any, ctx: InjectionContext) -> Any: + if ann is _HeaderAction: + return ctx.action + if ann is _HeaderUser: + return ctx.user + if ann is _HeaderExtra: + return ctx.extra bare = self._unwrap_optional(ann) if self._matches_state_annotation(bare, "pre"): return ctx.previous_state @@ -532,6 +584,18 @@ class CallbackRegistry: return type(None) in args and any(arg is str for arg in args) +class _HeaderAction: + """Sentinel annotation injecting the transaction action.""" + + +class _HeaderUser: + """Sentinel annotation injecting the transaction user.""" + + +class _HeaderExtra: + """Sentinel annotation injecting the display-only extra metadata.""" + + class _Unresolved: pass diff --git a/kanta/kanta.py b/kanta/kanta.py index 963f308..93eb19b 100644 --- a/kanta/kanta.py +++ b/kanta/kanta.py @@ -5,7 +5,7 @@ import logging from datetime import datetime from pathlib import Path from types import ModuleType, SimpleNamespace -from typing import Generic, TypeVar +from typing import Any, Generic, TypeVar from kanta.kantaimpl import KantaImpl from kanta.serialization import JsonSerializer, Serializer @@ -292,13 +292,38 @@ class Kanta(Generic[T]): return _register return _register(fn) + def logheader(self, fn=None): + """Register a transaction log header callback. + + Can be used as ``@kanta.logheader``. The callback formats the entire + header line printed before a transaction diff. It may declare + ``action: str``, ``user: str | None`` and ``extra: dict | None`` + parameters, and can also have ``DictPre``/``DictPost`` state dicts and + the ``Kanta`` instance injected. It must return ``str`` (or ``None`` + to fall through to the next callback, then to the default header). + + If registered, this replaces the default ``action by user`` header. + The ``extra`` metadata passed to :meth:`transaction` is display-only + and is never persisted; when no ``target`` key is supplied it defaults + to the database filename. + """ + + def _register(callback): + self._impl.add_logheader(callback) + return callback + + if fn is None: + return _register + return _register(fn) + def transaction( self, action: str, *, user: str | None = None, + extra: str | dict[str, Any] | None = None, mtime: bool | datetime = True, - log: bool | logging.Logger = True, + log: bool | logging.Logger | dict[str, bool] = True, ): """Create a transactional mutation context manager. @@ -307,6 +332,11 @@ class Kanta(Generic[T]): user: Optional user identifier stored in metadata and rendered in the log header. Register a ``@kanta.logfmt`` callback to format the user value; the path ``"$user"`` is passed for this case. + extra: Optional display-only metadata used for logging; it is not + persisted in the change record. A string is appended after + the action in the default header. A dict is passed to a + registered ``@kanta.logheader`` callback; if it has no + ``"target"`` key, the database filename is used. mtime: Controls the modification time ``m``. ``True`` (default) sets ``m`` to the current UTC time. ``False`` omits ``m`` so the previous modification time remains in effect; this is used for @@ -316,7 +346,9 @@ class Kanta(Generic[T]): log: Controls transaction logging. ``True`` (default) uses the ``kanta.transaction`` logger. ``False`` suppresses the transaction log. A :class:`~logging.Logger` instance writes - output to that logger instead. + output to that logger instead. A dict such as + ``{"header": True, "diff": False}`` toggles the header and + diff parts independently. Returns: A context manager yielding the live state object for mutation. @@ -330,6 +362,7 @@ class Kanta(Generic[T]): self._impl, action, user=user, + extra=extra, mtime=mtime, log=log, ) diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index ea6495a..f2dfeda 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -6,6 +6,7 @@ import asyncio import copy import importlib import logging +from collections.abc import Callable from datetime import UTC, datetime from types import SimpleNamespace from typing import Any, Generic, TypeVar @@ -80,6 +81,46 @@ class KantaImpl(PersistenceMixin, Generic[T]): """Register one migration logging callback.""" self.callback_registry.register("logmigr", callback) + def add_logheader(self, callback) -> None: + """Register one transaction header formatting callback.""" + self.callback_registry.register("logheader", callback) + + def build_headerfmt( + self, + action: str, + user: str | None, + extra: str | dict[str, Any] | None, + previous: dict | None, + current: dict | None, + ) -> tuple[Callable[..., str | None] | None, str | dict[str, Any] | None]: + """Build a headerfmt callable and normalized extra for ``log_change``. + + Returns ``(None, extra)`` unchanged when no logheader callback is + registered. Otherwise the extra dict gets a default ``target`` (the + database filename) when not supplied, so single-database apps get a + useful header with no extra code. + """ + if not self.callback_registry.has("logheader"): + return None, extra + if extra is None: + extra = {} + if isinstance(extra, dict) and "target" not in extra: + extra = {**extra, "target": self.filename.name} + ctx = InjectionContext( + action=action, + user=user, + extra=extra, + previous_state=previous, + current_state=current, + kanta=self._kanta, + ) + registry = self.callback_registry + + def headerfmt(action: str, user: str | None, extra: Any) -> str | None: + return registry.resolve_logheader(ctx) + + return headerfmt, extra + async def _handle_migration_log( self, migration_result: MigrationResult, @@ -303,12 +344,17 @@ class KantaImpl(PersistenceMixin, Generic[T]): resolved = logfmt(formatted_user, _USER_PATH) if resolved is not None: formatted_user = resolved + headerfmt, extra = self.build_headerfmt( + self.bootstrap_action, formatted_user, None, {}, current + ) log_change( self.bootstrap_action, record.diff, formatted_user, previous={}, + extra=extra, logfmt=logfmt, + headerfmt=headerfmt, logger=logger, level=logging.INFO, ) diff --git a/kanta/logging.py b/kanta/logging.py index b4ed8f4..38d9b59 100644 --- a/kanta/logging.py +++ b/kanta/logging.py @@ -33,6 +33,9 @@ _DELETE = "\033[1;31m" # Red for deletions _ADD = "\033[0;32m" # Green for additions _ACTION = "\033[1;34m" # Bold blue for action name _USER = "\033[0;34m" # Blue for user display +_ACTOR = "\033[0;36m" # Cyan for actor/label header fields +_SESSION = "\033[38;5;226m" # Bright yellow for session/request ids +_TARGET = "\033[38;5;250m" # White for target object names/ids # Metadata path used when formatting the transaction actor. _USER_PATH = "$user" @@ -116,17 +119,19 @@ def _format_path_components( def _format_path( - path: list[str], logfmt: Callable[[Any, str], str | None] | None + path: list[str], + logfmt: Callable[[Any, str], str | None] | None, + final_color: str = _PATH_FINAL, ) -> str: - """Format a path as dot notation with prefix in dark grey, final in default.""" + """Format a path as dot notation with prefix in dark grey, final colored.""" components = _format_path_components(path, logfmt) if not components: return "" if len(components) == 1: - return f"{_PATH_FINAL}{components[0]}{_RESET}" + return f"{final_color}{components[0]}{_RESET}" prefix = ".".join(components[:-1]) final = components[-1] - return f"{_PATH_PREFIX}{prefix}.{_RESET}{_PATH_FINAL}{final}{_RESET}" + return f"{_PATH_PREFIX}{prefix}.{_RESET}{final_color}{final}{_RESET}" def _get_nested(data: dict | None, path: list[str]) -> Any: @@ -202,8 +207,6 @@ def _format_change_lines( logfmt: Callable[[Any, str], str | None] | None = None, ) -> list[str]: """Format a single change as one or more lines.""" - path_str = _format_path(path, logfmt=logfmt) - if change_type == "delete": components = _format_path_components(path, logfmt) if len(components) == 1: @@ -213,6 +216,7 @@ def _format_change_lines( return [f" {_PATH_PREFIX}{prefix}.{_RESET}{_DELETE}{final} ✗{_RESET}"] if change_type == "add": + path_str = _format_path(path, logfmt, final_color=_ADD) if isinstance(value, dict) and value: lines = [f" {path_str} {_SEP}={_RESET}"] formatted_items = [] @@ -232,6 +236,7 @@ def _format_change_lines( return [f" {path_str} {_SEP}={_RESET} {value_str}"] value_str = _format_value(value, ".".join(path), logfmt=logfmt) + path_str = _format_path(path, logfmt=logfmt) return [f" {path_str} {_SEP}={_RESET} {value_str}"] @@ -262,9 +267,20 @@ def format_diff( return lines -def format_action_header(action: str, user: str | None = None) -> str: - """Format the action header line.""" +def format_action_header( + action: str, + user: str | None = None, + extra: str | dict[str, Any] | None = None, +) -> str: + """Format the action header line. + + A string *extra* is appended literally after the action; a dict *extra* + is ignored by the default header (it is meant for ``headerfmt`` + callbacks). + """ action_str = f"{_ACTION}{action}{_RESET}" + if isinstance(extra, str) and extra: + action_str = f"{action_str} {extra}" if user: user_str = f"{_USER}{user}{_RESET}" return f"{action_str} by {user_str}" @@ -276,10 +292,14 @@ def log_change( diff: dict, user: str | None = None, previous: dict | None = None, + extra: str | dict[str, Any] | None = None, logfmt: Callable[[Any, str], str | None] | None = None, + headerfmt: Callable[[str, str | None, Any], str | None] | None = None, *, logger: logging.Logger = transaction_logger, level: int = logging.INFO, + log_header: bool = True, + log_diff: bool = True, ) -> None: """Log a database change with pretty-printed diff. @@ -288,12 +308,31 @@ def log_change( diff: The JSON diff dict. user: Optional already-formatted user name to show in the header. previous: The previous state dict (for determining add vs update). + extra: Optional display-only metadata. A string is appended after + the action in the default header; a dict is passed to + ``headerfmt``. logfmt: Optional formatter callable ``(value, path) -> str | None``. + headerfmt: Optional header formatter callable + ``(action, user, extra) -> str | None`` replacing the default + header. Returning ``None`` falls back to the default header. logger: Logger to write to. Defaults to the ``kanta.transaction`` logger. level: Log level to use. Defaults to ``logging.INFO``. + log_header: Whether to emit the header line. + log_diff: Whether to emit the diff lines. """ - header = format_action_header(action, user) - diff_lines = format_diff(diff, previous, logfmt) + header: str | None = None + if log_header: + if headerfmt is not None: + header = headerfmt(action, user, extra) + if header is None: + header = format_action_header(action, user, extra) + + diff_lines = format_diff(diff, previous, logfmt) if log_diff else [] + + if header is None: + for line in diff_lines: + logger.log(level, line) + return if not diff_lines: logger.log(level, header) diff --git a/kanta/transaction.py b/kanta/transaction.py index 80b4803..0457ec6 100644 --- a/kanta/transaction.py +++ b/kanta/transaction.py @@ -5,6 +5,7 @@ from __future__ import annotations import logging from contextlib import contextmanager from datetime import datetime +from typing import Any from kanta.diff import compute_diff from kanta.exceptions import DataIntegrityError @@ -21,8 +22,9 @@ def transaction( action: str, *, user: str | None = None, + extra: str | dict[str, Any] | None = None, mtime: bool | datetime = True, - log: bool | logging.Logger = True, + log: bool | logging.Logger | dict[str, bool] = True, ): """Wrap writes in a transaction and yield the live db object.""" if impl.readonly: @@ -82,14 +84,29 @@ def transaction( if resolved is not None: formatted_user = resolved if log is not False: - logger = log if isinstance(log, logging.Logger) else transaction_logger + if isinstance(log, dict): + log_header = bool(log.get("header", True)) + log_diff = bool(log.get("diff", True)) + logger = transaction_logger + else: + log_header = log_diff = True + logger = ( + log if isinstance(log, logging.Logger) else transaction_logger + ) + headerfmt, extra = impl.build_headerfmt( + action, formatted_user, extra, previous, new_dict + ) log_change( action, record.diff, formatted_user, previous, - logfmt, + extra=extra, + logfmt=logfmt, + headerfmt=headerfmt, logger=logger, + log_header=log_header, + log_diff=log_diff, ) except Exception: _logger.warning("Transaction '%s' failed, rolling back changes", action)