FastAPI Cheatsheet 10 — Observability: structlog, OTEL, Prometheus

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.StackInfoRenderer(),
        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()

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

uv add opentelemetry-api opentelemetry-sdk \
   opentelemetry-exporter-otlp-proto-grpc \
   opentelemetry-instrumentation-fastapi \
   opentelemetry-instrumentation-httpx \
   opentelemetry-instrumentation-asyncpg \
   opentelemetry-instrumentation-sqlalchemy \
   opentelemetry-instrumentation-redis

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

FastAPIInstrumentor.instrument_app(app)
HTTPXClientInstrumentor().instrument()
AsyncPGInstrumentor().instrument()
SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine)

tracer = trace.get_tracer(__name__)

async def checkout(uid: int):
    with tracer.start_as_current_span("checkout") as span:
        span.set_attribute("user_id", str(uid))
        with tracer.start_as_current_span("validate_cart"):
            cart = await load_cart(uid)
        with tracer.start_as_current_span("charge"):
            payment = await charge_user(uid, cart.total)
            span.set_attribute("payment.id", payment.id)
        return payment

def add_trace_ids(_, __, 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

# Insert into structlog processors

from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

REQ = Counter("http_requests_total", "Total requests", ["method", "path", "status"])
LAT = Histogram("http_request_duration_seconds", "Latency", ["method", "path"])

@app.middleware("http")
async def metrics(request: Request, call_next):
    start = time.time()
    resp = await call_next(request)
    path = request.scope.get("route").path if request.scope.get("route") else request.url.path
    REQ.labels(request.method, path, resp.status_code).inc()
    LAT.labels(request.method, path).observe(time.time() - start)
    return resp

@app.get("/metrics", include_in_schema=False)
async def metrics_endpoint():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

from prometheus_fastapi_instrumentator import Instrumentator
Instrumentator().instrument(app).expose(app)

@app.get("/healthz", include_in_schema=False)
async def health(): return {"status": "ok"}

@app.get("/ready", include_in_schema=False)
async def ready(db: AsyncSession = Depends(get_db)):
    try:
        await db.execute(text("SELECT 1"))
    except Exception:
        return JSONResponse({"status": "not ready"}, status_code=503)
    return {"status": "ready"}

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

sentry_sdk.init(
    dsn=settings.sentry_dsn,
    integrations=[FastApiIntegration()],
    traces_sample_rate=0.1,
    environment=settings.env,
    release=settings.version,
)

@event.listens_for(engine.sync_engine, "before_cursor_execute")
def b(conn, cur, st, p, ctx, em): ctx._t = time.time()

@event.listens_for(engine.sync_engine, "after_cursor_execute")
def a(conn, cur, st, p, ctx, em):
    d = time.time() - ctx._t
    if d > 0.5:
        log.warning("slow_query", sql=st[:300], duration_ms=d * 1000)