Kanta colors log header parts: extra is a plain string, logheader callbacks receive pre-colored parts

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 CallbackRegistry._unwrap_optional(ann) is not str:
-        if name == "action" and bare is str:
+            return None
-            return _HeaderAction
+        return {
-        if name == "user" and bare is str:
+            "action": _HeaderAction,
-            return _HeaderUser
+            "user": _HeaderUser,
-        if name == "extra" and (bare is dict or get_origin(bare) is dict):
+            "extra": _HeaderExtra,
-            return _HeaderExtra
+        }.get(name)
-        return None
 
     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']")

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
+        Can be used as ``@kanta.logheader``.  The callback composes the header
-        header line printed before a transaction diff.  It may declare
+        line printed before a transaction diff from the parts it declares:
-        ``action: str``, ``user: str | None`` and ``extra: dict | None``
+        ``action: str``, ``user: str`` and ``extra: str``.  Kanta applies its
-        parameters, and can also have ``DictPre``/``DictPost`` state dicts and
+        header colors to the parts before calling the callback, and a missing
-        the ``Kanta`` instance injected.  It must return ``str`` (or ``None``
+        ``user``/``extra`` is passed as an empty string, so callbacks only
-        to fall through to the next callback, then to the default header).
+        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
+            extra: Optional display-only string appended after the action in
-                persisted in the change record.  A string is appended after
+                the log header (colored by Kanta), or passed to a registered
-                the action in the default header.  A dict is passed to a
+                ``@kanta.logheader`` callback.  It is never persisted in the
-                registered ``@kanta.logheader`` callback; if it has no
+                change record.
-                ``"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

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]:
+    ) -> Callable[..., str | None] | None:
-        """Build a headerfmt callable and normalized extra for ``log_change``.
+        """Build a headerfmt callable for ``log_change``.
 
-        Returns ``(None, extra)`` unchanged when no logheader callback is
+        The logheader callbacks receive the header parts with Kanta's colors
-        registered.  Otherwise the extra dict gets a default ``target`` (the
+        already applied (missing user/extra as empty strings).  Returns
-        database filename) when not supplied, so single-database apps get a
+        ``None`` when no logheader callback is registered.
-        useful header with no extra code.
         """
         if not self.callback_registry.has("logheader"):
-            return None, extra
+            return None
-        if extra is None:
+        action_str, user_str, extra_str = colorize_header_parts(action, user, extra)
-            extra = {}
-        if isinstance(extra, dict) and "target" not in extra:
-            extra = {**extra, "target": self.filename.name}
         ctx = InjectionContext(
-            action=action,
+            action=action_str,
-            user=user,
+            user=user_str,
-            extra=extra,
+            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,

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
+_TARGET = "\033[38;5;250m"  # White for the extra/target display
-_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"
@@ -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.
+    """Format the default action header line."""
-
+    action_str, user_str, extra_str = colorize_header_parts(action, user, extra)
-    A string *extra* is appended literally after the action; a dict *extra*
+    header = action_str
-    is ignored by the default header (it is meant for ``headerfmt``
+    if extra_str:
-    callbacks).
+        header = f"{header} {extra_str}"
-    """
+    if user_str:
-    action_str = f"{_ACTION}{action}{_RESET}"
+        header = f"{header} by {user_str}"
-    if isinstance(extra, str) and extra:
+    return header
-        action_str = f"{action_str} {extra}"
-    if user:
-        user_str = f"{_USER}{user}{_RESET}"
-        return f"{action_str} by {user_str}"
-    return action_str
 
 
 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
+        extra: Optional display-only string appended after the action in the
-            the action in the default header; a dict is passed to
+            default header (colored by Kanta), or passed to ``headerfmt``.
-            ``headerfmt``.
         logfmt: Optional formatter callable ``(value, path) -> str | None``.
-        headerfmt: Optional header formatter callable
+        headerfmt: Optional header formatter callable receiving the
-            ``(action, user, extra) -> str | None`` replacing the default
+            pre-colored ``(action, user, extra)`` parts and returning the
-            header.  Returning ``None`` falls back to the default header.
+            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.

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(

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']}",
+        headerfmt=lambda action, user, extra: f"CUSTOM {action} {extra}",
-        extra={"id": 7},
+        extra="7",
     )
     captured = capsys.readouterr()
     assert "CUSTOM update 7" in captured.err

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:
+    def header(action: str, user: str, extra: str) -> str:
-        return f"HDR {action} user={user} session={extra['session']}"
+        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):
+async def test_logheader_parts_are_colored_by_kanta(tmp_path, format_config, capsys):
-    caplog.set_level(logging.INFO, logger="kanta.transaction")
+    logging.getLogger("kanta").handlers.clear()
-    kanta = make_kanta(tmp_path / "mydb.db", Data, format_config)
+    configure_logging()
+    kanta = make_kanta(tmp_path / "test.db", Data, format_config)
 
     @kanta.logheader
-    def header(action: str, extra: dict | None) -> str:
+    def header(action: str, user: str, extra: str) -> str:
-        return f"target={extra['target']}"
+        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
+    assert "<><>" 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
 
 
 @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:
+    def second(action: str, extra: str) -> str:
-        return f"SECOND {action}"
+        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:
+    def header(action: str) -> str:
-        return f"BOOT {action} target={extra['target']}"
+        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(
+    with kanta.transaction(action="update", user="alice", extra="session-3") as data:
-        action="update", user="alice", extra={"session": 3, "target": "X"}
-    ) as data:
         data.counter = 1
     await kanta.close()