AskAiml
Sign in Get started

Log analyser CLI

A typer-based CLI with dataclasses, regex parsing, pathlib, and packaging via uv — end-to-end fundamentals.

🛠 Beginner

Goal

Build a fast command-line tool that parses, filters, and summarises log files. Output as table, JSON, or CSV. Distribute via PyPI.

This project lets you exercise idiomatic Python end-to-end: regex parsing, dataclasses, generators, pathlib, typer CLI, packaging with uv, full pytest suite, and CI.

Why this project

Recruiters give weight to "CLI on PyPI" because it shows: you can publish, structure, test, and document. Logs are ubiquitous; the tool is genuinely useful.

User experience

powershell
logq parse access.log
# Shows a table of (timestamp, method, path, status, latency_ms)

logq parse access.log --since 1h --status 500 --top 10
# Last hour, only 5xx, top 10 by latency

logq summary access.log --by status
# Counts grouped by status code

logq parse access.log --format json | jq .

cat access.log | logq parse - --format csv

Tech stack

  • typer (CLI) + rich (tables) + pydantic (config + model).
  • re + dataclasses for log lines.
  • pytest + hypothesis for tests.
  • ruff + mypy --strict + pre-commit.
  • uv for builds; GitHub Actions for CI + release; PyPI for publishing.

Architecture

src/logq/
├── __init__.py
├── __main__.py            # entry for `python -m logq`
├── cli.py                 # typer app
├── models.py              # LogRecord dataclass
├── parsers/
│   ├── __init__.py
│   ├── base.py            # Parser ABC
│   ├── nginx.py           # nginx combined log format
│   └── json_lines.py      # JSON-line logs
├── filters.py             # since, level, status, regex
├── formatters/
│   ├── table.py           # rich.Table
│   ├── json_out.py
│   └── csv_out.py
└── utils.py
tests/
├── conftest.py            # fixtures: sample logs
├── test_parsers.py
├── test_filters.py
├── test_cli.py            # typer.testing.CliRunner
└── test_properties.py     # hypothesis

Spec

LogRecord

python
from dataclasses import dataclass
from datetime import datetime

@dataclass(slots=True, frozen=True)
class LogRecord:
    ts: datetime
    method: str
    path: str
    status: int
    latency_ms: float | None
    bytes_sent: int | None
    raw: str

Parser interface

python
from typing import Protocol, Iterator

class Parser(Protocol):
    def parse(self, lines: Iterator[str]) -> Iterator[LogRecord]: ...

Nginx combined format

127.0.0.1 - - [10/Oct/2026:13:55:36 -0700] "GET /index.html HTTP/1.1" 200 2326 "-" "Mozilla/5.0"

Filters

  • --since <duration> (e.g. 1h, 30m, 2d).
  • --until <duration>.
  • --status <code> (multiple OK).
  • --method GET,POST.
  • --path-regex <pattern>.
  • --min-latency <ms>.

Output formats

  • table (default): rich table with colour by status (2xx green, 4xx yellow, 5xx red).
  • json: array of records.
  • csv: standard CSV.

Summary mode (logq summary)

  • --by status → counts per status.
  • --by path --top 10 → top 10 paths by request count.
  • --by hour → requests per hour bucket.

Acceptance criteria

  1. logq --help shows commands; logq parse --help shows options.
  2. Parses 100k nginx lines in under 1s on a modern laptop.
  3. Stream-parses stdin (no full in-memory buffering).
  4. Test coverage > 90%.
  5. mypy --strict passes.
  6. Published to TestPyPI; install via pip install -i https://test.pypi.org/simple/ logq works.
  7. CI runs on Linux/macOS/Windows × Python 3.10–3.13.

Stretch goals

  • logq tail -f mode (watch file).
  • Plugin system: third-party packages can register their own parsers via entry points.
  • --cache flag: parse once, store in SQLite, query fast.
  • Export to ClickHouse / DuckDB for ad-hoc analytics.
  • HTML report mode (Jinja2 + chart.js).
  • Anomaly detection on latency / status (z-score).

Key implementation hints

Streaming parser

python
def parse_file(path: Path | None) -> Iterator[LogRecord]:
    if path is None:
        source = sys.stdin
    else:
        source = path.open(encoding="utf-8", errors="replace")
    with source as f:
        yield from parser.parse(f)

Duration parsing

python
import re
from datetime import timedelta

_DUR = re.compile(r"^(\d+)\s*(s|m|h|d|w)$")
_UNITS = {"s": "seconds", "m": "minutes", "h": "hours", "d": "days", "w": "weeks"}

def parse_duration(s: str) -> timedelta:
    m = _DUR.match(s.strip())
    if not m:
        raise ValueError(f"invalid duration: {s!r}")
    return timedelta(**{_UNITS[m.group(2)]: int(m.group(1))})

Typer command

python
import typer
from pathlib import Path

app = typer.Typer(help="Parse and summarise logs.")

@app.command()
def parse(
    file: Path = typer.Argument(..., exists=True, allow_dash=True),
    since: str | None = typer.Option(None),
    status: list[int] = typer.Option(None),
    format: str = typer.Option("table"),
) -> None:
    records = parse_file(None if str(file) == "-" else file)
    if since:
        cutoff = datetime.now() - parse_duration(since)
        records = (r for r in records if r.ts >= cutoff)
    if status:
        records = (r for r in records if r.status in status)
    write(records, fmt=format)

if __name__ == "__main__":
    app()

Property-based tests

python
from hypothesis import given, strategies as st

@given(st.text(min_size=1, max_size=50))
def test_parser_doesnt_crash(line: str) -> None:
    list(NginxParser().parse(iter([line])))  # never raises

Deliverables

  • GitHub repo with README badges (CI, coverage, PyPI version).
  • Tagged release on PyPI.
  • 5-minute Loom demo (optional but recruiter-friendly).
  • Blog post explaining a non-trivial design decision (e.g., "why streaming over list[LogRecord]").

Lessons exercised

  • 01_fundamentals (all)
  • 02_oop (dataclasses, protocols)
  • 03_advanced (decorators, generators)
  • 04_stdlib_and_modern (regex, stdlib, asyncio for tail)
  • 06_testing (pytest, hypothesis, mypy)
  • 09_packaging (all)

Total estimated time: 6–10 hours for the core spec, 15–25 hours with stretch goals.