DEV Community: aidless

How I shipped structured JSON logging + Prometheus metrics with zero new dependencies

aidless — Mon, 13 Jul 2026 15:50:27 +0000

How I shipped structured JSON logging + Prometheus metrics with zero new dependencies

I almost added structlog and prometheus_client to my pyproject.toml. Then I read what they actually do.

Both libraries are excellent. structlog is the right call when you have a 30-engineer team shipping 50 services. prometheus_client is the right call when you have five teams of consumers scraping different metrics. For a single-author Python project with one process and one user, both are over-engineered. The 80 lines of code I would have pulled in, I can write in 200. The result: zero new runtime dependencies, full control over the output, and a smaller pip install footprint for every user.

Here is what I did instead.

The minimum useful observability surface

A small Python service needs four things, in order of importance:

Every log line is one JSON object. (No parsing for downstream tools.)
Every request has a trace id. Every log line in that request carries the same trace id. (So you can grep by id and see the whole story.)
Every log line goes to stderr. (So journald, Docker, and kubectl logs all see it without any extra configuration.)
Every metric is exposed in Prometheus text format at a stable URL.

structlog gives you #1, #2, #3 with a lot of flexibility. prometheus_client gives you #4 with a lot of flexibility. Both are about 16 MB of transitive dependencies combined. For a service that runs in a single process and exports maybe 20 metric names, the libraries are doing more work than the project.

The 80-line JsonFormatter

The custom logging formatter is the simplest part. The whole thing is here:

import json
import logging
from contextvars import ContextVar
from datetime import datetime, timezone

_trace_id_var: ContextVar[str | None] = ContextVar("trace_id", default=None)


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.now(tz=timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
        }
        tid = _trace_id_var.get()
        if tid:
            payload["trace_id"] = tid
        # Merge any extra={...} kwargs the logger call passed in.
        for k, v in record.__dict__.items():
            if k.startswith("_") or k in payload:
                continue
            if k in ("args", "msg", "levelname", "levelno", "pathname",
                    "filename", "module", "exc_info", "exc_text",
                    "stack_info", "lineno", "funcName", "created",
                    "msecs", "relativeCreated", "thread", "threadName",
                    "processName", "process", "message", "asctime"):
                continue
            try:
                json.dumps(v)
                payload[k] = v
            except TypeError:
                payload[k] = repr(v)
        if record.exc_info:
            payload["exc_info"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)


def configure_logging(level: str = "INFO") -> None:
    handler = logging.StreamHandler()
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers = [handler]
    root.setLevel(level)

Every log call becomes JSON. To attach structured fields, use logger.info("...", extra={"request_id": "abc", "tokens": 412}). The formatter merges them into the output. Three lines of filter logic keep the noise out (LogRecord internals, unjsonable values).

Trace IDs that survive `asyncio.to_thread`

The interesting part is the trace id. Python's ContextVar propagates across asyncio boundaries — including into threads spawned by asyncio.to_thread, because the context is copied at the task boundary. So:

@app.middleware("http")
async def observability_middleware(request, call_next):
    set_trace_id(request.headers.get("x-trace-id") or new_trace_id())
    t0 = time.perf_counter()
    try:
        response = await call_next(request)
        return response
    finally:
        m.inc_request(
            path=_collapse_path_params(request.url.path),
            method=request.method,
            status=response.status_code,
        )
        set_trace_id(None)  # don't leak to the next request

The middleware sets the trace id at the start of every request. Every log line emitted during that request — including from background tasks, from asyncio.to_thread workers, from anywhere — picks up the same id. After the request, we clear it so the next request on the same thread starts fresh.

The corresponding set_trace_id("...") is a one-liner:

def set_trace_id(trace_id: str | None) -> None:
    _trace_id_var.set(trace_id)

That's it. No middleware in every function. No thread-local hacks. The contextvar is propagated automatically.

A 200-line Prometheus store

For metrics, the same logic applies. prometheus_client is a 16 MB package that gives you Counter, Histogram, Gauge, and a registry. I need exactly those three types. Here is the whole class:

from collections import defaultdict
from dataclasses import dataclass, field
import threading


@dataclass
class _Histogram:
    buckets: list[float] = field(default_factory=lambda: [
        0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0,
        2.5, 5.0, 10.0, 30.0, 60.0,
    ])
    counts: list[int] = field(default_factory=lambda: [0] * 13)
    total: int = 0
    sum: float = 0.0

    def observe(self, value: float) -> None:
        self.total += 1
        self.sum += value
        for i, ub in enumerate(self.buckets):
            if value <= ub:
                self.counts[i] += 1


class Metrics:
    def __init__(self) -> None:
        self._lock = threading.Lock()
        self._counters: dict[tuple[str, frozenset], int] = defaultdict(int)
        self._histograms: dict[tuple[str, frozenset], _Histogram] = {}

    def inc_request(self, *, path, method, status) -> None:
        with self._lock:
            self._counters[("llm_lab_requests_total",
                            frozenset({"path": path, "method": method,
                                       "status": str(status).__class__(status)}))] += 1

A real implementation has more counters, but the structure is the same. Each counter is a key in a dict, indexed by (metric_name, frozen_label_set). The lock is held only for the increment; reads are lock-free because we always replace the integer atomically in CPython.

The render function is the part that matters most to operators. It produces the Prometheus text format directly:

# HELP llm_lab_requests_total Total HTTP requests
# TYPE llm_lab_requests_total counter
llm_lab_requests_total{method="POST",path="/submit",status="200"} 1
llm_lab_requests_total{method="GET",path="/metrics",status="200"} 1
# HELP llm_lab_request_duration_seconds HTTP request duration
# TYPE llm_lab_request_duration_seconds histogram
llm_lab_request_duration_seconds_bucket{le="0.005",method="POST",path="/submit"} 0
llm_lab_request_duration_seconds_bucket{le="0.01",method="POST",path="/submit"} 0
llm_lab_request_duration_seconds_count{method="POST",path="/submit"} 1
llm_lab_request_duration_seconds_sum{method="POST",path="/submit"} 0.012

This is the format Prometheus scrapers expect. We feed it from GET /metrics and call it done.

The label cardinality discipline

The most important thing this small store forces on you: label cardinality. With prometheus_client, you can declare Counter("foo", "label1") and pass any string at increment time, with no bound. That is how you accidentally create 10 million time series and get your scraper banned by Prometheus.

My Metrics class does not enforce this, but the call sites do. path is collapsed before use:

def _collapse_path_params(path: str) -> str:
    parts = path.strip("/").split("/")
    out = []
    for p in parts:
        if len(p) >= 8 and all(c in "0123456789abcdefABCDEF" for c in p):
            out.append(":id")
        else:
            out.append(p)
    return "/" + "/".join(out)

This replaces any path segment that looks like a 12-hex id with :id. So /result/abc123def456 becomes /result/:id. Every request to /result/abc... now collapses to the same series. Cardinality is bounded by code, not by data. This single design choice is the difference between a metrics store that scales and one that takes down your monitoring.

When you should actually use `structlog` and `prometheus_client`

The first 200 lines I wrote are enough for a single-service project that exports a few dozen metrics and doesn't have 30 engineers shipping it. You should reach for the libraries when:

You need structlog: distributed tracing (OpenTelemetry), log shipping to ELK / Loki, or the processor pipeline (cryptographic signing, redacting, sampling). The ContextVar + custom formatter pattern won't get you that.
You need prometheus_client: pull-based gauges (memory / CPU / queue depth), exemplars (linking a metric data point to a specific trace), multi-process mode (worker pool with shared memory), or exposition formats beyond the text format (protobuf, OpenMetrics).

If you don't need any of those, you're paying the libraries' tax without getting the benefit. For a small service, the cost is 16 MB of transitive dependencies, an import-time penalty of about 50ms, and an API surface that hides your project's actual observability shape behind a layer of generic abstraction.

What I have

For llm-lab — a single-process LLM evaluation framework — the 280 lines of observability.py give me:

One JSON object per log line, with trace id + any structured fields I want
Three metric families: HTTP requests (counter + histogram), LLM calls (counter + histogram), tokens (counter, by direction)
A GET /metrics endpoint that scrapers consume natively
A Metrics.snapshot() for tests, so I can assert "after this operation, the llm_lab_tokens_total{provider="openai",direction="prompt"} counter incremented by exactly 412"

The full code is in llm_lab/observability.py on GitHub. It's 280 lines including comments, type hints, and a render function. It depends on ContextVar (stdlib) and dataclasses (stdlib). Nothing else.

The same pattern would work for any service with maybe 20-50 metric series, one or two background workers, and a request flow that needs to be traceable end-to-end. If your service fits that description, try this approach first. You can always reach for structlog and prometheus_client later when the requirements actually demand it.

For AI

aidless — Sat, 16 May 2026 05:52:20 +0000

I'm a software engineering student about to graduate. Like many juniors, I was worried about the job market. So instead of waiting, I decided to build things — real AI projects that solve real problems.

The first project: an AI-powered code review agent that scans your codebase and flags bugs, security issues, and style problems. All in under 200 lines of Python.

Here's how I built it, what I learned, and why you should try it too.

The Stack

Layer	Tech
Language	Python 3.11
AI Model	DeepSeek V4 Pro (via Anthropic SDK)
Package Manager	uv
Output	Markdown report

Why DeepSeek instead of Claude directly? It's cheaper and the Anthropic-compatible endpoint makes migration trivial. The same code works with Claude by changing one line.

Architecture: 3 Simple Modules

code_review_agent/
├── scanner.py    # Walk directories, find code files
├── reviewer.py   # Send each file to the AI
├── report.py     # Generate Markdown report
└── main.py       # Wire everything together

No fancy frameworks. No LangChain. Just three functions chained together. I believe in starting simple and adding complexity only when needed.

1. Scanner — Finding the Code

The scanner walks a directory tree and collects every file the AI can review:

CODE_EXTENSIONS = {
    ".py": "Python", ".js": "JavaScript", ".ts": "TypeScript",
    ".java": "Java", ".go": "Go", ".rs": "Rust", ".sql": "SQL",
    # ... 10+ more
}

def scan_directory(root: str) -> list[FileInfo]:
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fname in filenames:
            ext = Path(fname).suffix.lower()
            if ext in CODE_EXTENSIONS:
                files.append(FileInfo(...))
    return files

Key decisions:

Skip dirs like .git, node_modules, .venv — no need to review third-party code
200KB size limit — don't burn API credits on generated files
Sort by language then path — makes the report predictable

2. Reviewer — The AI Brain

This is where the magic happens. Each file gets sent to the AI with a carefully designed system prompt:

REVIEW_SYSTEM_PROMPT = """\
You are a senior code reviewer. Analyze the code below and report issues.

For each issue you find, use EXACTLY this format:
  SEVERITY|LINE|CATEGORY|TITLE|DESCRIPTION

SEVERITY: critical, warning, info
CATEGORY: bug, security, performance, style, best-practice
LINE: approximate line number
"""

The pipe-delimited format is intentional. It's easy to parse, easy for the AI to follow, and human-readable at a glance.

What I learned about prompt design:

Be specific about output format — "use EXACTLY this format" prevents rambling
Give concrete examples — the AI follows patterns better than instructions
"Only report real issues" — without this, the AI invents problems to seem helpful

3. Report — Making It Readable

The report module takes parsed issues and generates clean Markdown. Issues are grouped by severity: critical first, then warnings, then suggestions. Each issue links back to the file and line number.

Running It

# Review any codebase
uv run python -m code_review_agent.main /path/to/project

# Review itself (demo mode)
uv run python -m code_review_agent.main

The First Run — and a Surprise

I ran it on its own source code. It found 4 issues in pyproject.toml:

"anthropic>=0.102.0 doesn't exist on PyPI" — Wrong. Version 0.102.0 exists.
"python-dotenv>=1.2.2 doesn't exist" — Also wrong. I literally installed it two hours ago.
"Missing langchain dependency" — Technically true, but I intentionally removed it.

Lesson: AI code review is a second pair of eyes, not a judge. It catches things you miss, but it also hallucinates. Always verify.

What's Next

This is project 2 of 4 in my ai-agent-playground series:

Hello Agent — First contact with the API
Code Review Agent — You just read about it
RAG Q&A System — Upload PDFs, ask questions with citations (coming soon)
Multi-Agent Crew — PM → Developer → QA → DevOps collaboration

Try It Yourself

git clone https://github.com/aidless/ai-agent-playground.git
cd ai-agent-playground
cp .env.example .env  # Add your API key
uv sync
uv run python -m code_review_agent.main /path/to/your/project

I'm documenting my journey from software engineering student to AI application developer. If you're on a similar path, let's connect — I'll be posting weekly updates here and on GitHub.

DEV Community: aidless

How I shipped structured JSON logging + Prometheus metrics with zero new dependencies

The minimum useful observability surface

The 80-line JsonFormatter

Trace IDs that survive asyncio.to_thread

A 200-line Prometheus store

The label cardinality discipline

When you should actually use structlog and prometheus_client

What I have

For AI

The Stack

Architecture: 3 Simple Modules

1. Scanner — Finding the Code

2. Reviewer — The AI Brain

3. Report — Making It Readable

Running It

The First Run — and a Surprise

What's Next

Try It Yourself

Trace IDs that survive `asyncio.to_thread`

When you should actually use `structlog` and `prometheus_client`