From 2c7487de73becf896415d861b45cdcd13865dc85 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Mon, 28 Sep 2026 04:50:06 +0000 Subject: [PATCH] Show the grep-matched records of a snapshot's state, change-log style --- kanta/__main__.py | 47 ++++++++++++++---- kanta/grep.py | 99 ++++++++++++++++++++++++++++++++------ tests/test_grep.py | 115 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 238 insertions(+), 23 deletions(-) diff --git a/kanta/__main__.py b/kanta/__main__.py index 6c77406..e57b760 100644 --- a/kanta/__main__.py +++ b/kanta/__main__.py @@ -18,7 +18,7 @@ import msgspec from kanta import Kanta from kanta.callbacks import InjectionContext, callback_error_reporter from kanta.exceptions import DatabaseError, DataIntegrityError, ReplayError -from kanta.grep import GrepPattern, evaluate, matches_snapshot +from kanta.grep import GrepPattern, evaluate, evaluate_snapshot, snapshot_record_lines from kanta.logging import ( LogEvent, emit_event, @@ -423,6 +423,21 @@ def _print_snapshot_indicator( _print(f"{label} {ts} {line}") +def _print_snapshot_matches( + state: dict[str, Any], kanta: Kanta[Any], highlight: Any +) -> None: + """Print the grep-matched records of a snapshot's state, change-log style. + + Each matched record is one ``path.to.record = {compact value}`` line + with normal path coloring and the matched regions highlighted. + """ + logfmt = kanta._impl.callback_registry.build_logfmt( + InjectionContext(kanta=kanta, previous_state=state, current_state=state) + ) + for line in snapshot_record_lines(state, highlight, logfmt): + _print(line) + + async def _log_migration( kanta: Kanta[Any], filename: Path, @@ -553,13 +568,24 @@ async def _run(args: argparse.Namespace) -> int: state = snap_event.snap.state version = snap_event.snap.v if not args.quiet: - _print_snapshot_indicator( - record_label(snap_event.line_number, snap_event.record_index), - snap_event.snap, - snapshot_line_to_index[snap_event.line_number], - kanta._impl.serializer, - ) - _print() + highlight = None + if grep_patterns: + logfmt = kanta._impl.callback_registry.build_logfmt( + InjectionContext( + kanta=kanta, previous_state=state, current_state=state + ) + ) + highlight = evaluate_snapshot(state, grep_patterns, logfmt=logfmt) + if not grep_patterns or highlight is not None: + _print_snapshot_indicator( + record_label(snap_event.line_number, snap_event.record_index), + snap_event.snap, + snapshot_line_to_index[snap_event.line_number], + kanta._impl.serializer, + ) + if highlight is not None: + _print_snapshot_matches(state, kanta, highlight) + _print() else: # Replay, collecting the item pool and tracking which items # survive the range window and --grep. With --context the pool @@ -598,9 +624,10 @@ async def _run(args: argparse.Namespace) -> int: current_state=current, ) ) - is_match = matches_snapshot( + highlight = evaluate_snapshot( current, grep_patterns, logfmt=logfmt ) + is_match = highlight is not None else: assert previous is not None if grep_patterns: @@ -657,6 +684,8 @@ async def _run(args: argparse.Namespace) -> int: snapshot_line_to_index[event.line_number], kanta._impl.serializer, ) + if highlight is not None: + _print_snapshot_matches(current, kanta, highlight) else: assert previous is not None _print_change_log( diff --git a/kanta/grep.py b/kanta/grep.py index 439fd00..4e9624f 100644 --- a/kanta/grep.py +++ b/kanta/grep.py @@ -1,9 +1,11 @@ -"""Structural grep matching and match highlighting for change records. +"""Structural grep matching and match highlighting for the kanta CLI. Backs the ``--grep`` option of the ``python -m kanta`` CLI. Patterns match against the structure of a transaction — its action, user, and the -dotted paths and values of its diff — never against rendered output -text. Matching is case-insensitive. +dotted paths and values of its diff — or against a snapshot's flattened +state, never against rendered output text. Matching is +case-insensitive. A matched change record is printed in full; a matched +snapshot shows only the matched lines of its state, rendered diff-style. Path terms match element-wise: each dotted element of the term must match a whole path element (``users`` matches ``users`` anywhere in the path @@ -32,10 +34,16 @@ import json from collections.abc import Iterator from typing import Any -from kanta.logging import _USER_PATH, _collect_changes, _get_nested +from kanta.logging import ( + _USER_PATH, + _collect_changes, + _format_path, + _format_value, + _get_nested, +) from kanta.serialization.base import unmarshal from kanta.structs import ChangeRecord -from kanta.tty import ANSI_RE, ESC, strip_ansi +from kanta.tty import ANSI_RE, ESC, Line, strip_ansi _GLOB_CHARS = frozenset("*?[") @@ -114,6 +122,31 @@ def _find_spans(text: str, needle: str) -> list[tuple[int, int]]: return spans +def snapshot_record_lines( + state: dict, highlighter: GrepHighlighter, logfmt: Any = None +) -> list[str]: + """Render the records of a snapshot state that matched grep patterns. + + Each matched entry is shown as one ``path.to.record = {compact value}`` + line, where the record is the top-level key plus one subkey — mirroring + change-record rendering, with normal path coloring, logfmt-prettified + elements, and the matched regions highlighted. + """ + records: dict[tuple[str, ...], None] = {} + for path in highlighter.matched_paths: + record = tuple(path[:2]) if len(path) > 1 else tuple(path) + records.setdefault(record) + lines = [] + for record in records: + path = list(record) + dotted = ".".join(path) + value = _get_nested(state, path) + path_str = _format_path(path, logfmt, highlight=highlighter) + value_str = _format_value(value, dotted, logfmt=logfmt, highlight=highlighter) + lines.append(str(Line()(" ", path_str, " ").sep("=")(" ", value_str))) + return lines + + @dataclasses.dataclass(frozen=True) class GrepPattern: """One parsed ``--grep`` pattern. @@ -306,6 +339,7 @@ class GrepHighlighter: self._lit_deletes: set[str] = set() self._value_marks: dict[str, list[_ValueMark]] = {} self._meta_marks: dict[str, list[_ValueMark]] = {} + self.matched_paths: list[list[str]] = [] def _light_elements(self, path: list[str], indices) -> None: for i in indices: @@ -318,6 +352,7 @@ class GrepHighlighter: vmark: _ValueMark | None, ) -> None: """Record one entry's match: lit element indices and/or a value mark.""" + self.matched_paths.append(entry.path) marked = vmark is not None and (vmark.needle or vmark.whole) if entry.anchor is None: if elements: @@ -376,6 +411,27 @@ class GrepHighlighter: return self._apply_marks(text, self._meta_marks.get(field, [])) +def _match_path_dual(term: str, path: list[str], logfmt: Any) -> frozenset[int] | None: + """Match a path term against raw and logfmt-prettified elements. + + A logfmt formatter may display an element differently (e.g. resolving a + user id to a name); the term matches either form, and the matched + element indices are lit so the displayed text gets marked. + """ + elements = _match_path(term, path) + if logfmt is not None: + pretty = [] + for i, element in enumerate(path): + resolved = logfmt(element, ".".join(path[: i + 1])) + pretty.append(resolved if resolved is not None else element) + pretty_elements = _match_path(term, pretty) + if pretty_elements is not None: + elements = ( + pretty_elements if elements is None else elements | pretty_elements + ) + return elements + + def _match_entry( pattern: GrepPattern, entry: _Entry, logfmt: Any ) -> tuple[frozenset[int] | None, _ValueMark | None] | None: @@ -389,33 +445,48 @@ def _match_entry( if logfmt is not None: pretty = logfmt(entry.raw, ".".join(entry.path)) if pattern.term is not None: - elements = _match_path(pattern.term, entry.path) + elements = _match_path_dual(pattern.term, entry.path, logfmt) vmark = _dual_mark(pattern.term, entry.raw, entry.text, pretty) if elements is None and vmark is None: return None else: - elements = _match_path(pattern.path_term or "", entry.path) + elements = _match_path_dual(pattern.path_term or "", entry.path, logfmt) vmark = _dual_mark(pattern.value_term or "", entry.raw, entry.text, pretty) if elements is None or vmark is None: return None return elements, vmark -def matches_snapshot( +def evaluate_snapshot( state: dict, patterns: list[GrepPattern], logfmt: Any = None -) -> bool: - """Match *patterns* against a snapshot's full state. +) -> GrepHighlighter | None: + """Match *patterns* against a snapshot's full state, returning its matched regions. The state is flattened into the same ``(path, value)`` entries change records are matched against, so path and value matching semantics are identical; a snapshot simply has no action or user to match. Returns - whether every pattern matched somewhere in the state. + ``None`` when any pattern matches nowhere in the state. """ entries = list(_leaf_entries([], unmarshal(state), None)) + highlighter = GrepHighlighter() for pattern in patterns: - if not any(_match_entry(pattern, entry, logfmt) for entry in entries): - return False - return True + matched = False + for entry in entries: + match = _match_entry(pattern, entry, logfmt) + if match is None: + continue + matched = True + highlighter.add(entry, *match) + if not matched: + return None + return highlighter + + +def matches_snapshot( + state: dict, patterns: list[GrepPattern], logfmt: Any = None +) -> bool: + """Return whether every pattern matched somewhere in a snapshot's state.""" + return evaluate_snapshot(state, patterns, logfmt) is not None def evaluate( diff --git a/tests/test_grep.py b/tests/test_grep.py index e22fe38..4362a7b 100644 --- a/tests/test_grep.py +++ b/tests/test_grep.py @@ -10,8 +10,10 @@ from kanta.grep import ( _match_path, _match_value, evaluate, + evaluate_snapshot, mark_spans, matches_snapshot, + snapshot_record_lines, ) from kanta.logging import _USER_PATH from kanta.serialization import JsonSerializer @@ -251,6 +253,41 @@ def test_matches_snapshot_supports_logfmt_prettified_values(): assert not matches_snapshot(state, _patterns("2026")) +def test_snapshot_record_lines_render_matched_records(): + """Matched entries render as one compact record line, change-log style.""" + state = { + "users": { + "alice": {"email": "alice@example.com", "age": 30}, + "bob": {"email": "bob@example.com"}, + }, + "counter": 1, + } + highlighter = evaluate_snapshot(state, _patterns("=alice@example.com")) + assert highlighter is not None + lines = snapshot_record_lines(state, highlighter) + # The whole alice record is shown; bob and counter are not. + assert [strip_ansi(line) for line in lines] == [ + " users.alice = {email: alice@example.com, age: 30}" + ] + assert MARK + "alice@example.com" + UNMARK in lines[0] + + +def test_snapshot_record_lines_pretty_path_and_value(): + """A logfmt-prettified path element matches and is marked, like values.""" + state = {"users": {"userid002": {"name": "Bob", "role": "editor"}}} + + def logfmt(value, path): + return "Bob" if value == "userid002" else None + + highlighter = evaluate_snapshot(state, _patterns("bob"), logfmt=logfmt) + assert highlighter is not None + lines = snapshot_record_lines(state, highlighter, logfmt) + assert len(lines) == 1 + assert strip_ansi(lines[0]) == " users.Bob = {name: Bob, role: editor}" + # Both displayed Bobs — the prettified path element and the name value. + assert lines[0].count(MARK) == 2 + + def test_entry_dataclass_holds_anchor_for_deletes(): entry = _Entry(["a", "b"], 1, "1", anchor=["a"]) assert entry.anchor == ["a"] @@ -421,6 +458,84 @@ def test_cli_grep_snapshot_prints_only_when_state_matches( assert "snapshot s0" not in err +def test_cli_grep_snapshot_prints_matched_state(tmp_path, capsys, monkeypatch): + """A matched snapshot shows the matched part of its state, diff-style.""" + state = { + "users": { + "alice": {"email": "alice@example.com", "age": 30}, + "bob": {"email": "bob@example.com"}, + } + } + + err = _run_cli( + tmp_path, + capsys, + monkeypatch, + _sample_changes(), + "--grep", + "=alice@example.com", + state=state, + ) + assert "snapshot s0" in err + # The matched state line is printed under the snapshot indicator; + # unmatched parts of the state (bob) are not shown. + assert "alice@example.com" in err + assert "bob@example.com" not in err + + # The matched region carries the yellow highlight when color is on. + err = _run_cli( + tmp_path, + capsys, + monkeypatch, + _sample_changes(), + "--grep", + "=alice@example.com", + state=state, + color=True, + ) + assert "\x1b[48;5;220malice@example.com\x1b[49m" in err + + +def test_cli_grep_snapshot_selection_shortcut(tmp_path, capsys, monkeypatch): + """-r sN applies --grep like the replay path: matched state is shown.""" + state = { + "users": { + "alice": {"email": "alice@example.com", "age": 30}, + "bob": {"email": "bob@example.com"}, + } + } + + err = _run_cli( + tmp_path, + capsys, + monkeypatch, + _sample_changes(), + "-r", + "s0", + "--grep", + "=alice@example.com", + state=state, + ) + assert "snapshot s0" in err + assert "users.alice" in err + assert "alice@example.com" in err + assert "bob@example.com" not in err + + # A non-matching pattern suppresses the explicitly selected snapshot. + err = _run_cli( + tmp_path, + capsys, + monkeypatch, + _sample_changes(), + "-r", + "s0", + "--grep", + "zzz", + state=state, + ) + assert "snapshot s0" not in err + + def test_cli_grep_path_value_forms(tmp_path, capsys, monkeypatch): """The 'path=value', 'path=' and '=value' forms restrict the match side.""" err = _run_cli(