Add experimental kanta CLI for inspecting databases.

demo/demo.py

@@ -85,7 +85,9 @@ async def main() -> None:
                 data.total = 99
                 raise ValueError("simulated failure")
         except ValueError:
-            print(f"\nReset rolled back: {data.total=} (we can always read data without tx)\n")
+            print(
+                f"\nReset rolled back: {data.total=} (we can always read data without tx)\n"
+            )
 
         with kanta.transaction(action="delete", user="userid002") as data:
             del data.users["userid001"]

kanta/logging.py

@@ -124,19 +124,22 @@ class LogEvent(msgspec.Struct, kw_only=True):
 def emit_event(
     ev: LogEvent,
     handlers: Iterable[Callable[[LogEvent], Any]] = (),
+    *,
+    fallback: Callable[[LogEvent], None] | None = None,
 ) -> None:
     """Dispatch *ev* through registered logemit handlers.
 
     Each handler receives the event and may log it (or not) as it sees fit.
     A falsy return value stops the chain: the event is considered handled.
     A truthy return value passes the event — possibly modified — to the next
-    handler.  When all handlers pass, :func:`default_emit` renders the event
-    with the built-in formatting.
+    handler.  When all handlers pass, the *fallback* renders the event;
+    the default fallback is :func:`default_emit` with the built-in formatting.
 
     Logging must never break functionality: a crashing handler is reported
-    and the chain falls back to the built-in formatting, and a failure in
-    the built-in formatting itself is reported and swallowed.
+    and the chain falls back to the fallback rendering, and a failure in
+    the fallback itself is reported and swallowed.
     """
+    render = fallback if fallback is not None else default_emit
     try:
         for handler in handlers:
             try:
@@ -146,7 +149,7 @@ def emit_event(
                 break
             if not proceed:
                 return
-        default_emit(ev)
+        render(ev)
     except Exception:
         _logger.exception("failed to emit %s log event", ev.kind)
 
@@ -187,6 +190,11 @@ def _join_path(path: str, key: str) -> str:
     return f"{path}.{key}"
 
 
+def _dim_ellipsis() -> str:
+    """Return the truncation ellipsis in the palette's ellipsis color."""
+    return str(Line().ellipsis("…"))
+
+
 def _format_value(
     value: Any,
     path: str,
@@ -209,7 +217,7 @@ def _format_value(
     if isinstance(value, str):
         value = _UNSAFE_CHARS.sub("", value)
         if len(value) > max_len:
-            return value[: max_len - 3] + "..."
+            return value[: max_len - 1] + _dim_ellipsis()
         return value
     if isinstance(value, dict):
         if not value:
@@ -235,7 +243,7 @@ def _format_value(
         return "[" + ", ".join(parts) + "]"
     text = str(value)
     if len(text) > max_len:
-        text = text[: max_len - 3] + "..."
+        text = text[: max_len - 1] + _dim_ellipsis()
     return text
 
 
@@ -361,15 +369,21 @@ def _format_change_lines(
         path_str = _format_path(path, logfmt, final_color="add")
         if isinstance(value, dict) and value:
             lines = [str(Line()("  ", path_str, " ").sep("="))]
-            formatted_items = []
             base_path = ".".join(path)
-            for k, v in value.items():
+            keys = []
+            for k in value:
                 key_path = _join_path(base_path, str(k))
-                key_display = _format_value(k, key_path, max_len=30, logfmt=logfmt)
-                v_str = _format_value(v, key_path, max_len=30, logfmt=logfmt)
-                formatted_items.append((key_display, v_str))
-            field_width = max(displaywidth(k) for k, _ in formatted_items)
+                keys.append((k, _format_value(k, key_path, max_len=30, logfmt=logfmt)))
+            field_width = max(displaywidth(kd) for _, kd in keys)
             field_width = max(field_width, 12)
+            # Each item line is "    {key:{field_width}}: {value}"; budget the
+            # value so the whole line fits in 80 columns.
+            value_width = max(80 - 4 - field_width - 2, 20)
+            formatted_items = []
+            for (k, key_display), v in zip(keys, value.values()):
+                key_path = _join_path(base_path, str(k))
+                v_str = _format_value(v, key_path, max_len=value_width, logfmt=logfmt)
+                formatted_items.append((key_display, v_str))
             return lines + [
                 str(
                     Line()("    ", k).sep(":")(

kanta/tty.py

@@ -75,6 +75,7 @@ class Colors:
     path_final = "38;5;250"  # White for the final path element
     add = "32"  # Green for additions
     delete = "1;31"  # Bold red for deletions
+    ellipsis = "38;5;242"  # Dark grey for the truncation ellipsis
 
 
 colors = Colors()

pyproject.toml

@@ -21,6 +21,9 @@ dependencies = [
     "msgspec>=0.20.0",
 ]
 
+[project.scripts]
+kanta = "kanta.__main__:main"
+
 [project.optional-dependencies]
 bin = [
     "blake3>=1.0.8",

tests/test_kanta_integration.py

@@ -562,13 +562,12 @@ async def test_migration_with_changes_records_diff_and_snapshot(
 
     records = read_changes(path, format_config)
     migration_records = [r for r in records if r.a.startswith("migrate")]
-    assert len(migration_records) == 2
+    # The version migration and the msgspec normalization that follows it are
+    # grouped into a single migrate:vN record.
+    assert len(migration_records) == 1
     assert migration_records[0].a == "migrate:v1"
     assert migration_records[0].v == 1
-    assert migration_records[0].diff == {"counter": 2}
-    assert migration_records[1].a == "migrate:msgspec"
-    assert migration_records[1].v == 1
-    assert migration_records[1].diff == {"users": {}}
+    assert migration_records[0].diff == {"counter": 2, "users": {}}
 
     snap = read_last_snapshot(path, format_config)
     assert snap is not None