Python Cheatsheet 08 — Logging with structlog

uv add structlog

import logging, sys, structlog

logging.basicConfig(format="%(message)s", stream=sys.stdout, level=logging.INFO)

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer() if PROD else structlog.dev.ConsoleRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    cache_logger_on_first_use=True,
)

log = structlog.get_logger()

log.info("user_created", user_id=42, email="[email protected]")
log.warning("rate_limited", user_id=42, count=100)
log.error("payment_failed", order_id=1, error="timeout")
log.exception("unhandled")        # includes exc_info from current handler

{"timestamp":"2026-05-17T...","level":"info","event":"user_created","user_id":42,"email":"[email protected]"}

# Bind context for a chain
log_user = log.bind(user_id=42)
log_user.info("step_a")
log_user.info("step_b")
# both have user_id=42

# new = bind + reset
log2 = log.new(request_id="abc")

# At request start
structlog.contextvars.bind_contextvars(request_id=rid, user_id=str(user.id))

# Anywhere downstream
log.info("event")     # automatically includes request_id, user_id

# At request end
structlog.contextvars.clear_contextvars()

from opentelemetry import trace

def add_trace(_, __, event_dict):
    span = trace.get_current_span()
    ctx = span.get_span_context()
    if ctx.trace_id:
        event_dict["trace_id"] = format(ctx.trace_id, "032x")
        event_dict["span_id"] = format(ctx.span_id, "016x")
    return event_dict

# Add to processors before JSONRenderer

import re

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
PHONE = re.compile(r"\+?\d[\d\s()-]{7,}")

def redact(_, __, ed):
    for k, v in list(ed.items()):
        if isinstance(v, str):
            v = EMAIL.sub("[EMAIL]", v)
            v = PHONE.sub("[PHONE]", v)
            ed[k] = v
    return ed

structlog.configure(
    wrapper_class=structlog.make_filtering_bound_logger(logging.WARNING),
)
# Drops INFO and DEBUG

log.info("user", user=lambda: expensive_serialize(u))
# Only serializes if log is emitted

import logging

# Route stdlib logging through structlog
structlog.configure(
    processors=[
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),
)

# Configure root logging
formatter = structlog.stdlib.ProcessorFormatter(
    processor=structlog.processors.JSONRenderer(),
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.addHandler(handler)

import random

def sample(_, __, ed):
    if ed.get("level") == "info" and random.random() < 0.1:
        return ed
    elif ed.get("level") in ("warning", "error", "critical"):
        return ed
    raise structlog.DropEvent

@app.middleware("http")
async def log_middleware(request, call_next):
    rid = request.headers.get("x-request-id") or str(uuid.uuid4())
    structlog.contextvars.bind_contextvars(
        request_id=rid,
        path=request.url.path,
        method=request.method,
    )
    try:
        return await call_next(request)
    finally:
        structlog.contextvars.clear_contextvars()