FastAPI Textbook Ch. 11 — Observability: Logging, Tracing, and Metrics

import structlog
import logging

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 request_id_middleware(request: 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:
        response = await call_next(request)
        response.headers["x-request-id"] = rid
        return response
    finally:
        structlog.contextvars.clear_contextvars()

@app.middleware("http")
async def user_context_middleware(request: Request, call_next):
    if user := getattr(request.state, "user", None):
        structlog.contextvars.bind_contextvars(user_id=str(user.id), tenant_id=user.tenant_id)
    return await call_next(request)

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

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

FastAPIInstrumentor.instrument_app(app)
HTTPXClientInstrumentor().instrument()
AsyncPGInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

async def checkout(user_id: int):
    with tracer.start_as_current_span("checkout") as span:
        span.set_attribute("user_id", str(user_id))
        
        with tracer.start_as_current_span("validate_cart"):
            cart = await load_cart(user_id)
        
        with tracer.start_as_current_span("charge"):
            payment = await charge_user(user_id, 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

# Add to structlog processors

from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

REQUEST_COUNT = Counter("http_requests_total", "Total requests", ["method", "path", "status"])
REQUEST_LATENCY = Histogram("http_request_duration_seconds", "Request latency", ["method", "path"])

@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    
    path = request.scope.get("route").path if request.scope.get("route") else request.url.path
    REQUEST_COUNT.labels(request.method, path, response.status_code).inc()
    REQUEST_LATENCY.labels(request.method, path).observe(duration)
    
    return response

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

from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app)

LOGIN_ATTEMPTS = Counter("auth_login_attempts_total", "Login attempts", ["result"])

@app.post("/token")
async def login(...):
    try:
        user = await authenticate(...)
        LOGIN_ATTEMPTS.labels("success").inc()
    except InvalidCredentials:
        LOGIN_ATTEMPTS.labels("failure").inc()
        raise

@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,
)

async def audit_log(actor_id, action, target, before=None, after=None):
    await db.execute(
        insert(AuditLog).values(
            actor_id=actor_id, action=action, target=target,
            before=before, after=after, occurred_at=datetime.utcnow(),
        )
    )

@app.put("/users/{id}")
async def update_user(id: int, data: UserUpdate, current: User = Depends(get_current_user), db = Depends(get_db)):
    user = await db.get(User, id)
    before = user.dict()
    update_fields(user, data)
    after = user.dict()
    await audit_log(current.id, "user.update", {"type": "user", "id": id}, before, after)
    await db.commit()

def should_sample(request):
    if request.url.path == "/healthz": return False
    if random.random() < 0.1: return True
    return False