From e0725c57385b46a74b91004aed047d02712f40a2 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Fri, 7 Aug 2026 00:31:04 +0000 Subject: [PATCH] Kanta colors log header parts: extra is a plain string, logheader callbacks receive pre-colored parts --- kanta/callbacks.py | 19 ++++--- kanta/kanta.py | 32 ++++++------ kanta/kantaimpl.py | 41 +++++++-------- kanta/logging.py | 58 +++++++++++---------- kanta/transaction.py | 5 +- tests/test_logging.py | 30 +++++++++-- tests/test_logheader.py | 108 ++++++++++++++++++++++++---------------- 7 files changed, 173 insertions(+), 120 deletions(-) diff --git a/kanta/callbacks.py b/kanta/callbacks.py index bde0870..0770536 100644 --- a/kanta/callbacks.py +++ b/kanta/callbacks.py @@ -63,7 +63,7 @@ class InjectionContext: migration_result: MigrationResult | None = None action: str | None = None user: str | None = None - extra: Any = None + extra: str | None = None @dataclass @@ -245,14 +245,13 @@ class CallbackRegistry: @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 + if CallbackRegistry._unwrap_optional(ann) is not str: + return None + return { + "action": _HeaderAction, + "user": _HeaderUser, + "extra": _HeaderExtra, + }.get(name) def _validate_function( self, @@ -520,7 +519,7 @@ class CallbackRegistry: if kind == "logheader": parts.append("action: str") parts.append("user: str | None") - parts.append("extra: dict | None") + parts.append("extra: str | None") if kind in {"logfmt", "logheader"}: parts.append("Annotated[dict, 'pre']") parts.append("Annotated[dict, 'post']") diff --git a/kanta/kanta.py b/kanta/kanta.py index 73d152e..a7d9485 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 Any, Generic, TypeVar +from typing import Generic, TypeVar from kanta.kantaimpl import KantaImpl from kanta.serialization import JsonSerializer, Serializer @@ -316,17 +316,18 @@ class Kanta(Generic[T]): 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). + Can be used as ``@kanta.logheader``. The callback composes the header + line printed before a transaction diff from the parts it declares: + ``action: str``, ``user: str`` and ``extra: str``. Kanta applies its + header colors to the parts before calling the callback, and a missing + ``user``/``extra`` is passed as an empty string, so callbacks only + arrange text — no color codes, fallbacks, or padding. ``DictPre``/ + ``DictPost`` state dicts and the ``Kanta`` instance can also be + injected. The callback must be synchronous and 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): @@ -342,7 +343,7 @@ class Kanta(Generic[T]): action: str, *, user: str | None = None, - extra: str | dict[str, Any] | None = None, + extra: str | None = None, mtime: bool | datetime = True, log: bool | logging.Logger | dict[str, bool] = True, ): @@ -353,11 +354,10 @@ 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. + extra: Optional display-only string appended after the action in + the log header (colored by Kanta), or passed to a registered + ``@kanta.logheader`` callback. It is never persisted in the + change record. 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 diff --git a/kanta/kantaimpl.py b/kanta/kantaimpl.py index 93e5b75..a2f2682 100644 --- a/kanta/kantaimpl.py +++ b/kanta/kantaimpl.py @@ -13,7 +13,13 @@ from typing import Any, Generic, TypeVar from kanta.callbacks import CallbackRegistry, InjectionContext from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError -from kanta.logging import _USER_PATH, bootstrap_logger, log_change, migration_logger +from kanta.logging import ( + _USER_PATH, + bootstrap_logger, + colorize_header_parts, + log_change, + migration_logger, +) from kanta.migrations import MigrationResult, Migrations from kanta.persistence import PersistenceMixin from kanta.serialization import restore_data_in_place, struct_to_dict @@ -89,37 +95,33 @@ class KantaImpl(PersistenceMixin, Generic[T]): self, action: str, user: str | None, - extra: str | dict[str, Any] | None, + extra: str | 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``. + ) -> Callable[..., str | None] | None: + """Build a headerfmt callable 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. + The logheader callbacks receive the header parts with Kanta's colors + already applied (missing user/extra as empty strings). Returns + ``None`` when no logheader callback is registered. """ 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} + return None + action_str, user_str, extra_str = colorize_header_parts(action, user, extra) ctx = InjectionContext( - action=action, - user=user, - extra=extra, + action=action_str, + user=user_str, + extra=extra_str, previous_state=previous, current_state=current, kanta=self._kanta, ) registry = self.callback_registry - def headerfmt(action: str, user: str | None, extra: Any) -> str | None: + def headerfmt(action: str, user: str, extra: str) -> str | None: return registry.resolve_logheader(ctx) - return headerfmt, extra + return headerfmt async def _handle_migration_log( self, @@ -350,7 +352,7 @@ class KantaImpl(PersistenceMixin, Generic[T]): resolved = logfmt(formatted_user, _USER_PATH) if resolved is not None: formatted_user = resolved - headerfmt, extra = self.build_headerfmt( + headerfmt = self.build_headerfmt( self.bootstrap_action, formatted_user, None, {}, current ) log_change( @@ -358,7 +360,6 @@ class KantaImpl(PersistenceMixin, Generic[T]): record.diff, formatted_user, previous={}, - extra=extra, logfmt=logfmt, headerfmt=headerfmt, logger=logger, diff --git a/kanta/logging.py b/kanta/logging.py index 38d9b59..10f4cc3 100644 --- a/kanta/logging.py +++ b/kanta/logging.py @@ -33,9 +33,7 @@ _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 +_TARGET = "\033[38;5;250m" # White for the extra/target display # Metadata path used when formatting the transaction actor. _USER_PATH = "$user" @@ -267,24 +265,35 @@ def format_diff( return lines +def colorize_header_parts( + action: str, + user: str | None = None, + extra: str | None = None, +) -> tuple[str, str, str]: + """Apply Kanta's header colors to the action, user, and extra parts. + + ``None`` user/extra become empty strings so custom header callbacks can + interpolate the parts directly without fallbacks. + """ + action_str = f"{_ACTION}{action}{_RESET}" + user_str = f"{_USER}{user}{_RESET}" if user else "" + extra_str = f"{_TARGET}{extra}{_RESET}" if extra else "" + return action_str, user_str, extra_str + + def format_action_header( action: str, user: str | None = None, - extra: str | dict[str, Any] | None = None, + extra: str | 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}" - return action_str + """Format the default action header line.""" + action_str, user_str, extra_str = colorize_header_parts(action, user, extra) + header = action_str + if extra_str: + header = f"{header} {extra_str}" + if user_str: + header = f"{header} by {user_str}" + return header def log_change( @@ -292,9 +301,9 @@ def log_change( diff: dict, user: str | None = None, previous: dict | None = None, - extra: str | dict[str, Any] | None = None, + extra: str | None = None, logfmt: Callable[[Any, str], str | None] | None = None, - headerfmt: Callable[[str, str | None, Any], str | None] | None = None, + headerfmt: Callable[[str, str, str], str | None] | None = None, *, logger: logging.Logger = transaction_logger, level: int = logging.INFO, @@ -308,13 +317,12 @@ 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``. + extra: Optional display-only string appended after the action in the + default header (colored by Kanta), or 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. + headerfmt: Optional header formatter callable receiving the + pre-colored ``(action, user, extra)`` parts and returning the + header line. 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. diff --git a/kanta/transaction.py b/kanta/transaction.py index 79b416f..f0de37e 100644 --- a/kanta/transaction.py +++ b/kanta/transaction.py @@ -5,7 +5,6 @@ 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 @@ -22,7 +21,7 @@ def transaction( action: str, *, user: str | None = None, - extra: str | dict[str, Any] | None = None, + extra: str | None = None, mtime: bool | datetime = True, log: bool | logging.Logger | dict[str, bool] = True, ): @@ -95,7 +94,7 @@ def transaction( if isinstance(log, logging.Logger) else transaction_logger ) - headerfmt, extra = impl.build_headerfmt( + headerfmt = impl.build_headerfmt( action, formatted_user, extra, previous, new_dict ) log_change( diff --git a/tests/test_logging.py b/tests/test_logging.py index a019579..ad9e8d8 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -2,7 +2,29 @@ import logging import pytest -from kanta.logging import configure_logging, log_change +from kanta.logging import ( + _ACTION, + _RESET, + _TARGET, + _USER, + colorize_header_parts, + configure_logging, + log_change, +) + + +def test_colorize_header_parts(): + action, user, extra = colorize_header_parts("update", "alice", "tgt") + assert action == f"{_ACTION}update{_RESET}" + assert user == f"{_USER}alice{_RESET}" + assert extra == f"{_TARGET}tgt{_RESET}" + + +def test_colorize_header_parts_missing_user_and_extra(): + action, user, extra = colorize_header_parts("update") + assert action == f"{_ACTION}update{_RESET}" + assert user == "" + assert extra == "" @pytest.fixture(autouse=True) @@ -55,7 +77,7 @@ def test_log_change_appends_extra_string(capsys): log_change("export", {}, extra="mydb.db") captured = capsys.readouterr() assert "export" in captured.err - assert "mydb.db" in captured.err + assert f"{_TARGET}mydb.db{_RESET}" in captured.err def test_log_change_headerfmt_replaces_header(capsys): @@ -65,8 +87,8 @@ def test_log_change_headerfmt_replaces_header(capsys): log_change( "update", {}, - headerfmt=lambda action, user, extra: f"CUSTOM {action} {extra['id']}", - extra={"id": 7}, + headerfmt=lambda action, user, extra: f"CUSTOM {action} {extra}", + extra="7", ) captured = capsys.readouterr() assert "CUSTOM update 7" in captured.err diff --git a/tests/test_logheader.py b/tests/test_logheader.py index 1186593..bdbb6ca 100644 --- a/tests/test_logheader.py +++ b/tests/test_logheader.py @@ -4,10 +4,21 @@ import pytest from kanta import Kanta from kanta.callbacks import DictPost, DictPre +from kanta.logging import _ACTION, _RESET, _TARGET, _USER, configure_logging from .support import Data, make_kanta, read_changes +@pytest.fixture(autouse=True) +def _reset_kanta_loggers(): + yield + for name in ("kanta", "kanta.transaction", "kanta.bootstrap", "kanta.migration"): + logger = logging.getLogger(name) + logger.setLevel(logging.NOTSET) + logger.propagate = True + logger.handlers.clear() + + def test_logheader_rejects_async(tmp_path, format_config): kanta = make_kanta(tmp_path / "test.db", Data, format_config) @@ -38,59 +49,74 @@ def test_logheader_rejects_unknown_annotation(tmp_path, format_config): return action +def test_logheader_rejects_dict_extra(tmp_path, format_config): + kanta = make_kanta(tmp_path / "test.db", Data, format_config) + + with pytest.raises(TypeError, match="unsupported annotation"): + + @kanta.logheader + def header(action: str, extra: dict) -> str: + return action + + @pytest.mark.asyncio async def test_logheader_replaces_default_header(tmp_path, format_config, caplog): caplog.set_level(logging.INFO, logger="kanta.transaction") kanta = make_kanta(tmp_path / "test.db", Data, format_config) @kanta.logheader - def header(action: str, user: str | None, extra: dict | None) -> str: - return f"HDR {action} user={user} session={extra['session']}" + def header(action: str, user: str, extra: str) -> str: + return f"{user} {action} {extra}" await kanta.open(log=False) - with kanta.transaction(action="update", user="alice", extra={"session": 3}) as data: + with kanta.transaction(action="update", user="alice", extra="tgt") as data: data.counter = 1 await kanta.close() - assert "HDR update user=alice session=3" in caplog.text + # caplog strips ANSI codes; the colors are verified via capsys below. + assert "alice update tgt" in caplog.text # The diff body is still logged after the custom header. assert "counter" in caplog.text @pytest.mark.asyncio -async def test_logheader_default_target_is_filename(tmp_path, format_config, caplog): - caplog.set_level(logging.INFO, logger="kanta.transaction") - kanta = make_kanta(tmp_path / "mydb.db", Data, format_config) +async def test_logheader_parts_are_colored_by_kanta(tmp_path, format_config, capsys): + logging.getLogger("kanta").handlers.clear() + configure_logging() + kanta = make_kanta(tmp_path / "test.db", Data, format_config) @kanta.logheader - def header(action: str, extra: dict | None) -> str: - return f"target={extra['target']}" + def header(action: str, user: str, extra: str) -> str: + return f"{user} {action} {extra}" + + await kanta.open(log=False) + with kanta.transaction(action="update", user="alice", extra="tgt") as data: + data.counter = 1 + await kanta.close() + + err = capsys.readouterr().err + assert f"{_USER}alice{_RESET}" in err + assert f"{_ACTION}update{_RESET}" in err + assert f"{_TARGET}tgt{_RESET}" in err + + +@pytest.mark.asyncio +async def test_logheader_missing_user_and_extra_are_empty( + tmp_path, format_config, caplog +): + caplog.set_level(logging.INFO, logger="kanta.transaction") + kanta = make_kanta(tmp_path / "test.db", Data, format_config) + + @kanta.logheader + def header(action: str, user: str, extra: str) -> str: + return f"<{user}><{extra}>" await kanta.open(log=False) with kanta.transaction(action="update") as data: data.counter = 1 await kanta.close() - assert "target=mydb.db" in caplog.text - - -@pytest.mark.asyncio -async def test_logheader_explicit_target_kept(tmp_path, format_config, caplog): - caplog.set_level(logging.INFO, logger="kanta.transaction") - kanta = make_kanta(tmp_path / "mydb.db", Data, format_config) - - @kanta.logheader - def header(action: str, extra: dict | None) -> str: - return f"target={extra['target']}" - - await kanta.open(log=False) - with kanta.transaction( - action="update", extra={"target": "Project X (abcd1234)"} - ) as data: - data.counter = 1 - await kanta.close() - - assert "target=Project X (abcd1234)" in caplog.text + assert "<><>" in caplog.text @pytest.mark.asyncio @@ -106,7 +132,7 @@ async def test_logheader_injects_states_and_kanta(tmp_path, format_config, caplo kanta: Kanta, ) -> str: return ( - f"{action} counter {previous.get('counter')}" + f"counter {previous.get('counter')}" f" -> {current.get('counter')} db={kanta.filename.name}" ) @@ -115,7 +141,7 @@ async def test_logheader_injects_states_and_kanta(tmp_path, format_config, caplo data.counter = 5 await kanta.close() - assert "increment counter 0 -> 5 db=test.db" in caplog.text + assert "counter 0 -> 5 db=test.db" in caplog.text @pytest.mark.asyncio @@ -128,15 +154,15 @@ async def test_logheader_chain_first_non_none_wins(tmp_path, format_config, capl return None @kanta.logheader - def second(action: str) -> str: - return f"SECOND {action}" + def second(action: str, extra: str) -> str: + return f"SECOND {extra}" await kanta.open(log=False) - with kanta.transaction(action="update") as data: + with kanta.transaction(action="update", extra="marked") as data: data.counter = 1 await kanta.close() - assert "SECOND update" in caplog.text + assert "SECOND marked" in caplog.text @pytest.mark.asyncio @@ -169,7 +195,7 @@ async def test_logheader_receives_formatted_user(tmp_path, format_config, caplog return "Alice" @kanta.logheader - def header(action: str, user: str | None) -> str: + def header(action: str, user: str) -> str: return f"actor={user}" await kanta.open(log=False) @@ -186,13 +212,13 @@ async def test_logheader_applies_to_bootstrap(tmp_path, format_config, caplog): kanta = make_kanta(tmp_path / "test.db", Data, format_config) @kanta.logheader - def header(action: str, extra: dict | None) -> str: - return f"BOOT {action} target={extra['target']}" + def header(action: str) -> str: + return f"BOOT {action}" await kanta.open() await kanta.close() - assert "BOOT bootstrap target=test.db" in caplog.text + assert "BOOT bootstrap" in caplog.text @pytest.mark.asyncio @@ -217,9 +243,7 @@ async def test_extra_is_not_persisted(tmp_path, format_config): kanta = make_kanta(path, Data, format_config) await kanta.open(log=False) - with kanta.transaction( - action="update", user="alice", extra={"session": 3, "target": "X"} - ) as data: + with kanta.transaction(action="update", user="alice", extra="session-3") as data: data.counter = 1 await kanta.close()