Implement support for NO_COLOR/FORCE_COLOR env with isatty and journald checks for autodetection.

2026-09-12 22:07:37 +00:00
parent ab5b7584c9
commit 2e6f48bac5
7 changed files with 117 additions and 23 deletions
@@ -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); `.<colorname>` 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

@@ -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


@@ -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:
@@ -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.

@@ -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"
@@ -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 = []
@@ -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()