diff --git a/CHANGELOG.md b/CHANGELOG.md index 91a38b2..79a789c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to this project are documented in this file. ## Unreleased +- Phase 8, CLI, complete: + - `logquill tail [--level=] [--json] [-f/--follow] [-n/--lines]` — a + `logquill` console-script for tailing a JSONL log file in local dev. + Human-readable output by default, colorized by level to match + `ConsoleTransport`; `--json` prints each matching record as a raw JSON + line instead. `--level` filters to that level and above; `-n` limits to + the last N matching records; `-f`/`--follow` keeps polling the file for + newly appended records, for a `tail -f`-style live view. A line that + isn't valid JSON, or isn't a JSON object, is skipped with a warning on + stderr instead of aborting the whole tail. - Phase 7, advanced context & stdlib bridge, complete: - `bind_context(**values)` — a `contextvars`-based context manager that merges `values` into every `Logger` call underneath it, through any diff --git a/README.md b/README.md index 9bf3575..c0e1765 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ for what's landed so far. - **Zero required runtime dependencies** — stdlib only; `aiohttp` is opt-in, for async HTTP - **Typed throughout** — `mypy --strict` clean on the public API - **Context propagation, exception capture & the stdlib bridge** — `bind_context()` (`contextvars`-based, no manual passing), `exc_info=` on any `Logger` method (formatted traceback into `meta["stack"]`), `LogQuillHandler` (bridges stdlib `logging` into a `Logger`), and `RateLimitPlugin` — see [Context propagation, exception capture & the stdlib bridge](#context-propagation-exception-capture--the-stdlib-bridge) +- **CLI** — `logquill tail app.log --level=warn --json -f` for filtering/following a JSONL log file in local dev, no extra install — see [CLI](#cli) ## Install @@ -854,6 +855,32 @@ for _ in range(100): logger.error("connection refused") # only the first 5 per minute ship ``` +## CLI + +Installing `logquill` also installs a `logquill` command for local +development — no extra dependencies, since it only reads the JSONL files any +`FileTransport`/`ConsoleTransport` already writes: + +```bash +# print every record in the file, human-readable +logquill tail app.log + +# only WARN and above, as raw JSON lines +logquill tail app.log --level=warn --json + +# only the last 20 matching records +logquill tail app.log -n 20 + +# keep watching the file and print new records as they're appended, like `tail -f` +logquill tail app.log -f +``` + +Human-readable output is colorized by level (matching `ConsoleTransport`'s +colors) when writing to a terminal; pass `--no-color` to disable that, or +`--json` to print each matching record as a single JSON line instead. A line +that isn't valid JSON, or isn't a JSON object, is skipped with a warning on +stderr rather than aborting the whole tail. + ## Development ```bash diff --git a/logquill/cli.py b/logquill/cli.py new file mode 100644 index 0000000..05319d4 --- /dev/null +++ b/logquill/cli.py @@ -0,0 +1,240 @@ +"""The `logquill` command-line entry point (`pip install logquill` → `logquill tail ...`).""" + +from __future__ import annotations + +import argparse +import contextlib +import json +import sys +import time +from pathlib import Path +from typing import IO, Any, Sequence + +from logquill.levels import Level, parse_level + +_COLORS = { + Level.TRACE: "\x1b[90m", + Level.DEBUG: "\x1b[36m", + Level.INFO: "\x1b[32m", + Level.WARN: "\x1b[33m", + Level.ERROR: "\x1b[31m", + Level.FATAL: "\x1b[35m", +} +_RESET = "\x1b[0m" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="logquill", description="LogQuill command-line tools for local development." + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + tail_parser = subparsers.add_parser( + "tail", help="Print (and optionally follow) a LogQuill JSONL log file." + ) + tail_parser.add_argument("file", help="Path to a LogQuill JSONL log file.") + tail_parser.add_argument( + "--level", + default=None, + help="Only show records at or above this level (e.g. --level=error).", + ) + tail_parser.add_argument( + "--json", + action="store_true", + help="Print raw JSON lines instead of human-readable text.", + ) + tail_parser.add_argument( + "-f", + "--follow", + action="store_true", + help="Keep watching the file and print new records as they're appended.", + ) + tail_parser.add_argument( + "-n", + "--lines", + type=int, + default=None, + metavar="N", + help="Only show the last N matching records instead of the whole file.", + ) + tail_parser.add_argument( + "--no-color", + action="store_true", + help="Disable ANSI colorization, even when writing to a terminal.", + ) + return parser + + +def _passes_filter(record: dict[str, Any], min_level: Level | None) -> bool: + if min_level is None: + return True + try: + return parse_level(record.get("level")) >= min_level # type: ignore[arg-type] + except (TypeError, ValueError): + return False + + +def _parse_line(line: str, *, warn_stream: IO[str]) -> dict[str, Any] | None: + line = line.strip() + if not line: + return None + try: + parsed = json.loads(line) + except json.JSONDecodeError: + warn_stream.write(f"logquill tail: skipping malformed JSON line: {line[:200]!r}\n") + return None + if not isinstance(parsed, dict): + warn_stream.write(f"logquill tail: skipping non-object JSON line: {line[:200]!r}\n") + return None + return parsed + + +def _format_human(record: dict[str, Any], *, colorize: bool) -> str: + level_name = str(record.get("level", "?")) + timestamp = record.get("timestamp", "?") + logger_name = record.get("logger", "?") + message = record.get("message", "") + meta = record.get("meta") or {} + + line = f"{timestamp} {level_name:<5} {logger_name}: {message}" + if meta: + line += f" {json.dumps(meta, separators=(',', ':'), default=str)}" + + if colorize: + try: + color = _COLORS.get(parse_level(level_name)) + except (TypeError, ValueError): + color = None + if color: + line = f"{color}{line}{_RESET}" + return line + + +def _emit(record: dict[str, Any], *, as_json: bool, colorize: bool, out: IO[str]) -> None: + if as_json: + out.write(json.dumps(record, separators=(",", ":"), default=str) + "\n") + else: + out.write(_format_human(record, colorize=colorize) + "\n") + out.flush() + + +def _read_existing( + path: Path, + *, + min_level: Level | None, + lines: int | None, + as_json: bool, + colorize: bool, + out: IO[str], + warn_stream: IO[str], +) -> int: + """Prints every matching record currently in the file; returns the byte offset at EOF.""" + matched: list[dict[str, Any]] = [] + with path.open("r", encoding="utf-8") as f: + for raw_line in f: + record = _parse_line(raw_line, warn_stream=warn_stream) + if record is not None and _passes_filter(record, min_level): + matched.append(record) + offset = f.tell() + + if lines is not None: + matched = matched[-lines:] + for record in matched: + _emit(record, as_json=as_json, colorize=colorize, out=out) + return offset + + +def _follow( + path: Path, + offset: int, + *, + min_level: Level | None, + as_json: bool, + colorize: bool, + out: IO[str], + warn_stream: IO[str], + poll_interval: float = 0.5, + max_iterations: int | None = None, +) -> None: + """Polls `path` for lines appended after `offset`, forever unless `max_iterations` is set. + + A polling loop rather than an inotify/kqueue watch: it keeps this module + dependency-free and behaves the same across platforms, at the cost of up to + `poll_interval` seconds of latency on a new line — an acceptable trade for a + local dev tool. + """ + iterations = 0 + while max_iterations is None or iterations < max_iterations: + iterations += 1 + try: + with path.open("r", encoding="utf-8") as f: + f.seek(offset) + new_lines = f.readlines() + offset = f.tell() + except FileNotFoundError: + time.sleep(poll_interval) + continue + + for raw_line in new_lines: + record = _parse_line(raw_line, warn_stream=warn_stream) + if record is not None and _passes_filter(record, min_level): + _emit(record, as_json=as_json, colorize=colorize, out=out) + + time.sleep(poll_interval) + + +def _run_tail( + args: argparse.Namespace, + *, + min_level: Level | None, + out: IO[str], + warn_stream: IO[str], +) -> int: + path = Path(args.file) + if not path.exists(): + warn_stream.write(f"logquill tail: no such file: {args.file}\n") + return 1 + + colorize = not args.no_color and not args.json and getattr(out, "isatty", lambda: False)() + offset = _read_existing( + path, + min_level=min_level, + lines=args.lines, + as_json=args.json, + colorize=colorize, + out=out, + warn_stream=warn_stream, + ) + + if args.follow: + with contextlib.suppress(KeyboardInterrupt): + _follow( + path, + offset, + min_level=min_level, + as_json=args.json, + colorize=colorize, + out=out, + warn_stream=warn_stream, + ) + return 0 + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if args.command != "tail": + parser.error(f"Unknown command: {args.command}") + + try: + min_level = parse_level(args.level) if args.level is not None else None + except ValueError as exc: + parser.error(str(exc)) + return 2 # pragma: no cover - argparse.error() already exits + + return _run_tail(args, min_level=min_level, out=sys.stdout, warn_stream=sys.stderr) + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index b2f22c7..b865f33 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,6 +72,9 @@ hooks = [ "pre-commit>=3.7", ] +[project.scripts] +logquill = "logquill.cli:main" + [project.urls] Homepage = "https://github.com/nikhilvdev/logquill-python" Repository = "https://github.com/nikhilvdev/logquill-python" diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..5c44ba5 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import io +import json +from pathlib import Path + +import pytest + +from logquill.cli import _follow, _read_existing, _run_tail, build_parser, main +from logquill.levels import Level + + +def _write_lines(path: Path, records: list[dict]) -> None: + path.write_text("\n".join(json.dumps(r) for r in records) + "\n", encoding="utf-8") + + +def _tail(file: str, *extra_args: str, min_level: Level | None = None) -> tuple[str, str, int]: + args = build_parser().parse_args(["tail", file, *extra_args]) + out, warn = io.StringIO(), io.StringIO() + exit_code = _run_tail(args, min_level=min_level, out=out, warn_stream=warn) + return out.getvalue(), warn.getvalue(), exit_code + + +def test_tail_prints_human_readable_lines_by_default(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + _write_lines( + log_path, + [ + { + "timestamp": "2026-01-01T00:00:00.000Z", + "level": "INFO", + "logger": "app", + "message": "started", + "meta": {"pid": 1}, + } + ], + ) + + output, _warn, exit_code = _tail(str(log_path)) + + assert exit_code == 0 + assert "2026-01-01T00:00:00.000Z" in output + assert "INFO" in output + assert "app: started" in output + assert '{"pid":1}' in output + + +def test_tail_json_flag_prints_raw_json_lines(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + record = { + "timestamp": "2026-01-01T00:00:00.000Z", + "level": "WARN", + "logger": "app", + "message": "careful", + "meta": {}, + } + _write_lines(log_path, [record]) + + output, _warn, _exit_code = _tail(str(log_path), "--json") + + assert json.loads(output.strip()) == record + + +def test_tail_filters_by_level(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + _write_lines( + log_path, + [ + {"timestamp": "t", "level": "DEBUG", "logger": "app", "message": "quiet", "meta": {}}, + {"timestamp": "t", "level": "ERROR", "logger": "app", "message": "loud", "meta": {}}, + ], + ) + + output, _warn, _exit_code = _tail(str(log_path), "--level", "error", min_level=Level.ERROR) + + assert "loud" in output + assert "quiet" not in output + + +def test_tail_lines_flag_limits_to_last_n_matching_records(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + _write_lines( + log_path, + [ + {"timestamp": "t", "level": "INFO", "logger": "app", "message": f"line-{i}", "meta": {}} + for i in range(5) + ], + ) + + output, _warn, _exit_code = _tail(str(log_path), "-n", "2") + + assert "line-3" in output + assert "line-4" in output + assert "line-0" not in output + assert "line-1" not in output + assert "line-2" not in output + + +def test_tail_skips_malformed_lines_and_warns(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + log_path.write_text( + "not json\n" + + json.dumps( + {"timestamp": "t", "level": "INFO", "logger": "app", "message": "ok", "meta": {}} + ) + + "\n" + + json.dumps(["not", "an", "object"]) + + "\n", + encoding="utf-8", + ) + + output, warn, exit_code = _tail(str(log_path)) + + assert exit_code == 0 + assert "ok" in output + assert "malformed" in warn + assert "non-object" in warn + + +def test_tail_missing_file_returns_nonzero_and_warns(tmp_path: Path) -> None: + missing = tmp_path / "does-not-exist.log" + + _output, warn, exit_code = _tail(str(missing)) + + assert exit_code == 1 + assert "no such file" in warn + + +def test_tail_unknown_level_exits_with_usage_error( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + log_path = tmp_path / "app.log" + log_path.write_text("", encoding="utf-8") + + with pytest.raises(SystemExit) as exc_info: + main(["tail", str(log_path), "--level", "nonsense"]) + + assert exc_info.value.code == 2 + assert "Unknown log level" in capsys.readouterr().err + + +def test_tail_follow_picks_up_appended_records(tmp_path: Path) -> None: + log_path = tmp_path / "app.log" + _write_lines( + log_path, + [{"timestamp": "t", "level": "INFO", "logger": "app", "message": "first", "meta": {}}], + ) + + setup_out = io.StringIO() + offset = _read_existing( + log_path, + min_level=None, + lines=None, + as_json=False, + colorize=False, + out=setup_out, + warn_stream=io.StringIO(), + ) + + second = {"timestamp": "t", "level": "INFO", "logger": "app", "message": "second", "meta": {}} + with log_path.open("a", encoding="utf-8") as f: + f.write(json.dumps(second) + "\n") + + follow_out = io.StringIO() + _follow( + log_path, + offset, + min_level=None, + as_json=False, + colorize=False, + out=follow_out, + warn_stream=io.StringIO(), + poll_interval=0.01, + max_iterations=1, + ) + + assert "second" in follow_out.getvalue() + assert "first" not in follow_out.getvalue()