From 2e6f48bac549cb1c444d030b7255648d90e714eb Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Sat, 12 Sep 2026 22:07:37 +0000 Subject: [PATCH] Implement support for NO_COLOR/FORCE_COLOR env with isatty and journald checks for autodetection. --- docs/database.md | 3 ++- kanta/__main__.py | 39 ++++++++++++++++++++++++++------------- kanta/logging.py | 22 ++++++++++++++++------ kanta/tty.py | 19 +++++++++++++++++++ tests/test_cli.py | 28 +++++++++++++++++++++++++++- tests/test_logemit.py | 25 ++++++++++++++++++++++++- tests/test_logging.py | 4 +++- 7 files changed, 117 insertions(+), 23 deletions(-) diff --git a/docs/database.md b/docs/database.md index 3aac425..07673ec 100644 --- a/docs/database.md +++ b/docs/database.md @@ -171,7 +171,7 @@ def resolve_user_key(value: str) -> str | None: #### Transaction Log Headers -- By default a transaction is logged with an `action by user` header followed by the diff lines. Added paths are colored green, deleted paths red. +- By default a transaction is logged with an `action by user` header followed by the diff lines. Added paths are colored green, deleted paths red. ANSI color codes are stripped after formatting when the standard error stream does not support color: `NO_COLOR` disables colors, `FORCE_COLOR` forces them, otherwise a tty check and a journald (`JOURNAL_STREAM`) check decide. The CLI (`python -m kanta`) strips its output the same way. - `kanta.transaction(..., extra=...)` accepts a display-only value that is shown after the action in the header. Anything other than `None` is printed str-converted (colored by Kanta), unless a custom logemit handler does something else with it; it is never persisted in the `ChangeRecord`. - `kanta.transaction(..., logdiff=False)` skips building and printing the diff body and logs only the header, which is useful for large or noisy changesets. Diff output can also be disabled globally with `configure_logging(diff=False)`; diff lines are emitted on the `kanta.transaction.diff` child logger so applications can route or silence them separately from the headers. @@ -203,6 +203,7 @@ def emit(ev: LogEvent): - `colors`: the mutable color palette. Colors are bare SGR parameter strings (e.g. `"1;34"`, `"38;5;226"`) without escape framing. Attributes are read at render time, so assignments (`colors.action = "36"`) and additions (`colors.session = "38;5;226"`) take effect immediately. - `Line`: builds a terminal string part by part. Calling it appends content (`str`-converted); `.` arms a palette color for the next call only, and the reset is folded into a single escape sequence with whatever color comes next. `width=`/`align=` pad by display width; `str(line)` finishes the line and restores default colors. - `strip_ansi`, `displaywidth` (wide chars and emoji count correctly) and `pad` for working with pre-colored strings. + - `use_color(stream)`: the color-support test used by Kanta's own output — honors `NO_COLOR`/`FORCE_COLOR`, then `stream.isatty()`, then the journald `JOURNAL_STREAM` device/inode match. ## Migrations diff --git a/kanta/__main__.py b/kanta/__main__.py index 202a7de..2b698b2 100644 --- a/kanta/__main__.py +++ b/kanta/__main__.py @@ -33,7 +33,7 @@ from kanta.replaylog import ( ) from kanta.serialization import Serializer, dict_to_struct, struct_to_dict from kanta.structs import ChangeRecord, Snapshot -from kanta.tty import Line +from kanta.tty import Line, strip_ansi, use_color EXIT_SUCCESS = 0 EXIT_GENERIC = 1 @@ -45,6 +45,19 @@ EXIT_VALIDATION_ERROR = 21 _logger = logging.getLogger(__name__) +def _print(*args: Any) -> None: + """Print to stderr, stripping ANSI codes when the stream has no color support. + + Color detection runs per call so redirected or reassigned ``sys.stderr`` + (and environment changes) are honored; ANSI codes are stripped after + formatting, not by formatting differently. + """ + text = " ".join(str(arg) for arg in args) + if not use_color(): + text = strip_ansi(text) + print(text, file=sys.stderr) + + class _CliError(Exception): """A user-facing error message paired with a process exit code.""" @@ -234,14 +247,14 @@ def _print_change_log( ts = _format_ts(record.ts) lines = ev.diff_lines if not lines: - print(f"{label} {ts} {ev.header}", file=sys.stderr) + _print(f"{label} {ts} {ev.header}") elif len(lines) == 1: - print(f"{label} {ts} {ev.header}{lines[0]}", file=sys.stderr) + _print(f"{label} {ts} {ev.header}{lines[0]}") else: - print(f"{label} {ts} {ev.header}", file=sys.stderr) + _print(f"{label} {ts} {ev.header}") for line in lines: - print(line, file=sys.stderr) - print(file=sys.stderr) + _print(line) + _print() emit_event( event, @@ -319,7 +332,7 @@ def _print_snapshot_indicator( line.target(f" {_format_ts(snap.m)}") size = len(serializer.encode(snap.state)) line.path_prefix(f" {_format_size(size)}") - print(f"{label} {ts} {line}", file=sys.stderr) + _print(f"{label} {ts} {line}") async def _log_migration( @@ -359,7 +372,7 @@ async def _log_migration( migrations=descriptions, ), registry.logemit_handlers, - fallback=lambda ev: print(ev.header, file=sys.stderr), + fallback=lambda ev: _print(ev.header), ) @@ -457,7 +470,7 @@ async def _run(args: argparse.Namespace) -> int: snapshot_line_to_index[snap_event.line_number], kanta._impl.serializer, ) - print(file=sys.stderr) + _print() else: # Replay up to the range end, printing logs within the range. state = {} @@ -481,7 +494,7 @@ async def _run(args: argparse.Namespace) -> int: _print_change_log(label, event.record, previous, current, kanta) printed = True if printed: - print(file=sys.stderr) + _print() # Apply optional migrations to the range-end state. if kanta._impl.migrations is not None: @@ -518,7 +531,7 @@ async def _run(args: argparse.Namespace) -> int: # The file was already fully decoded and validated above with # the object's own serializer, and its migrations were applied # to the state; no need to re-open through a new instance. - print(f"{data}", file=sys.stderr) + _print(f"{data}") output_state = struct_to_dict(data, serializer=kanta._impl.serializer) else: kanta_typed = Kanta( @@ -526,7 +539,7 @@ async def _run(args: argparse.Namespace) -> int: ) try: await kanta_typed.open(create=False, readonly=True, log=False) - print(f"{data}", file=sys.stderr) + _print(f"{data}") except (msgspec.ValidationError, msgspec.DecodeError) as exc: raise _CliError( f"Validation error: {exc}", EXIT_VALIDATION_ERROR @@ -577,7 +590,7 @@ def main(argv: list[str] | None = None) -> int: try: return asyncio.run(_run(args)) except _CliError as exc: - print(exc, file=sys.stderr) + _print(exc) return exc.code diff --git a/kanta/logging.py b/kanta/logging.py index 2d9fff2..2386da0 100644 --- a/kanta/logging.py +++ b/kanta/logging.py @@ -5,6 +5,8 @@ through :func:`emit_event`, which runs any registered ``logemit`` callbacks and falls back to :func:`default_emit` for the built-in formatting. Diff output is formatted in a human-readable path notation style with color coding; see :mod:`kanta.tty` for the color palette and line builder. +ANSI codes are stripped at emit time when the standard error stream does +not support color (``NO_COLOR``/``FORCE_COLOR``, tty and journald checks). """ import logging @@ -16,7 +18,7 @@ from typing import Any import msgspec from kanta.serialization.base import _apply, unmarshal -from kanta.tty import Line, displaywidth +from kanta.tty import Line, displaywidth, strip_ansi, use_color transaction_logger = logging.getLogger("kanta.transaction") bootstrap_logger = logging.getLogger("kanta.bootstrap") @@ -155,6 +157,11 @@ def emit_event( _logger.exception("failed to emit %s log event", ev.kind) +def _maybe_strip(text: str) -> str: + """Strip ANSI codes from *text* when stderr has no color support.""" + return text if use_color() else strip_ansi(text) + + def default_emit(ev: LogEvent) -> None: """Emit *ev* with Kanta's built-in formatting. @@ -163,25 +170,28 @@ def default_emit(ev: LogEvent) -> None: logger so it can be silenced or routed separately from the headers. This is what runs when no logemit callback handles the event; custom callbacks may call it to delegate events they do not care about. + + ANSI color codes are stripped after formatting when the standard error + stream does not support color (see :func:`kanta.tty.use_color`). """ if ev.kind != "change": - ev.logger.log(ev.level, ev.header) + ev.logger.log(ev.level, _maybe_strip(ev.header)) return diff_logger = logging.getLogger(f"{ev.logger.name}.diff") lines = ev.diff_lines if ev.show_diff and diff_logger.isEnabledFor(ev.level) else [] if not lines: - ev.logger.log(ev.level, ev.header) + ev.logger.log(ev.level, _maybe_strip(ev.header)) return if len(lines) == 1: - diff_logger.log(ev.level, f"{ev.header}{lines[0]}") + diff_logger.log(ev.level, _maybe_strip(f"{ev.header}{lines[0]}")) return - ev.logger.log(ev.level, ev.header) + ev.logger.log(ev.level, _maybe_strip(ev.header)) for line in lines: - diff_logger.log(ev.level, line) + diff_logger.log(ev.level, _maybe_strip(line)) def _join_path(path: str, key: str) -> str: diff --git a/kanta/tty.py b/kanta/tty.py index 560de50..13d7e28 100644 --- a/kanta/tty.py +++ b/kanta/tty.py @@ -10,8 +10,12 @@ color instead of emitting a separate one. from __future__ import annotations +import io +import os import re +import sys import unicodedata +from contextlib import suppress from typing import Any ESC = "\x1b[" @@ -25,6 +29,21 @@ def strip_ansi(text: str) -> str: return ANSI_RE.sub("", text) +def use_color(stream: io.TextIOBase = sys.stderr) -> bool: + """Test if the stream supports color codes.""" + if os.environ.get("NO_COLOR"): # Non empty means no (no-color.org) + return False + if os.environ.get("FORCE_COLOR", "") not in {"", "0"}: # force-color.org, node + return True + if hasattr(stream, "isatty") and stream.isatty(): + return True + with suppress(KeyError, ValueError, OSError): # Journald does color (-ocat) + dev, ino = map(int, os.environ["JOURNAL_STREAM"].split(":", 1)) + st = os.fstat(stream.fileno()) + return st.st_dev == dev and st.st_ino == ino + return False + + def displaywidth(text: str) -> int: """Return the terminal column width of *text*, ignoring ANSI sequences. diff --git a/tests/test_cli.py b/tests/test_cli.py index 0399464..54f813f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -76,8 +76,10 @@ def test_extra_import_paths_ignores_other_python_versions(tmp_path, monkeypatch) assert str(other_site) not in sys.path -def test_cli_snapshot_line_format(tmp_path, capsys): +def test_cli_snapshot_line_format(tmp_path, capsys, monkeypatch): """Snapshot lines are timestamped and colored with metadata.""" + monkeypatch.setenv("FORCE_COLOR", "1") + monkeypatch.delenv("NO_COLOR", raising=False) path = tmp_path / "test.kantadb" ts = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC) mtime = datetime(2026, 8, 12, 9, 0, 0, tzinfo=UTC) @@ -105,6 +107,30 @@ def test_cli_snapshot_line_format(tmp_path, capsys): assert "\x1b[38;5;242m 13 B" in err +def test_cli_strips_ansi_without_color_support(tmp_path, capsys, monkeypatch): + """Without a tty and with NO_COLOR set, output contains no ANSI codes.""" + monkeypatch.setenv("NO_COLOR", "1") + monkeypatch.delenv("FORCE_COLOR", raising=False) + path = tmp_path / "test.kantadb" + ts = datetime(2026, 8, 12, 10, 6, 52, 375398, tzinfo=UTC) + serializer = JsonSerializer() + framer = LineFramer() + + snapshot = Snapshot(ts=ts, v=1, m=None, state={"counter": 5}) + change = ChangeRecord(ts=ts, a="inc", v=1, u="user1", diff={"counter": 6}) + data = framer.frame_snapshot( + serializer.encode(snapshot), record_offset=0 + ) + framer.frame_change(serializer.encode(change), record_offset=0) + path.write_bytes(data) + + code = main([str(path)]) + assert code == 0 + + err = capsys.readouterr().err + assert "\x1b[" not in err + assert "snapshot s0" in err + + def test_import_dotted_from_file_path(tmp_path): """--data can be a filesystem path with an optional colon-separated symbol.""" module = tmp_path / "models.py" diff --git a/tests/test_logemit.py b/tests/test_logemit.py index 97acfdb..84a14ac 100644 --- a/tests/test_logemit.py +++ b/tests/test_logemit.py @@ -127,6 +127,25 @@ def test_default_emit_created_and_migrated(capsys): assert "🛢️ x.kantadb migrated v0 -> v1: migrate_v1 (rename)" in err +def test_default_emit_strips_ansi_without_color_support(capsys, monkeypatch): + """NO_COLOR output contains no ANSI codes; FORCE_COLOR keeps them.""" + logging.getLogger("kanta").handlers.clear() + configure_logging() + monkeypatch.setenv("NO_COLOR", "1") + monkeypatch.delenv("FORCE_COLOR", raising=False) + emit_event(_change_event(diff={"counter": 1})) + err = capsys.readouterr().err + assert "\x1b[" not in err + assert "counter" in err + + logging.getLogger("kanta").handlers.clear() + configure_logging() + monkeypatch.setenv("FORCE_COLOR", "1") + monkeypatch.delenv("NO_COLOR", raising=False) + emit_event(_change_event(diff={"counter": 1})) + assert "\x1b[" in capsys.readouterr().err + + @pytest.mark.asyncio async def test_logemit_receives_transaction_events(tmp_path, format_config): path = tmp_path / "test.db" @@ -242,7 +261,11 @@ async def test_logmigr_failure_does_not_break_open(tmp_path, format_config): @pytest.mark.asyncio -async def test_aborted_transaction_emits_event(tmp_path, format_config, caplog): +async def test_aborted_transaction_emits_event( + tmp_path, format_config, caplog, monkeypatch +): + monkeypatch.setenv("FORCE_COLOR", "1") + monkeypatch.delenv("NO_COLOR", raising=False) path = tmp_path / "test.db" kanta = make_kanta(path, Data, format_config) events = [] diff --git a/tests/test_logging.py b/tests/test_logging.py index 52960c5..5582118 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -71,7 +71,9 @@ def test_log_change_no_diff(capsys): assert "test" in captured.err -def test_log_change_appends_extra_string(capsys): +def test_log_change_appends_extra_string(capsys, monkeypatch): + monkeypatch.setenv("FORCE_COLOR", "1") + monkeypatch.delenv("NO_COLOR", raising=False) kanta_logger = logging.getLogger("kanta") kanta_logger.handlers.clear() configure_logging()