Recap — Structured Logs, Request Context, Exception Handlers

# Structured Logs, Request Context, Exception Handlers For any FastAPI app in production, you need: structured logs, request tracing, clean error responses. Middleware is FastAPI's mechanism. ## Structured JSON logs ```python import logging import sys from pythonjsonlogger import jsonlogger def setup_logging(level: str = "INFO"): handler = logging.StreamHandler(sys.stdout) handler.setFormatter(jsonlogger.JsonFormatter( "%(asctime)s %(name)s %(levelname)s %(message)s" )) root = logging.getLogger() root.setLevel(level) root.addHandler(handler) # In main.py setup_logging("INFO") logger = logging.getLogger(__name__) logger.info("user_login", extra={"user_id": 123, "ip": "1.2.3.4"}) ``` Output: ```json {"asctime":"2026-05-27T14:32:01","name":"my_app.auth","levelname":"INFO","message":"user_login","user_id":123,"ip":"1.2.3.4"} ``` Queryable by log aggregators (Datadog, ELK, Loki). ## Request ID middleware Trace requests across logs: ```python import uuid from fastapi import Request from contextvars import ContextVar request_id_var: ContextVar[str] = ContextVar("request_id", default="") @app.middleware("http") async def add_request_id(request: Request, call_next): request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4()) request_id_var.set(request_id) response = await call_next(request) response.headers["X-Request-ID"] = request_id return response ``` In your log formatter, include the request_id: ```python class RequestIDFilter(logging.Filter): def filter(self, record): record.request_id = request_id_var.get() return True logger.addFilter(RequestIDFilter()) ``` Now all logs in a request share the same request_id. Trace failures end-to-end. ## Access log middleware ```python @app.middleware("http") async def log_requests(request: Request, call_next): start = time.perf_counter() response = await call_next(request) duration_ms = (time.perf_counter() - start) * 1000 logger.info( "request", extra={ "method": request.method, "path": request.url.path, "status": response.status_code, "duration_ms": round(duration_ms, 2), "client_ip": request.client.host if request.client else None, }, ) return response ``` Every request logged with method, path, status, duration. ## Exception handlers For custom exceptions in your app: ```python class AppError(Exception): """Base app exception.""" status_code: int = 500 code: str = "internal_error" class NotFoundError(AppError): status_code = 404 code = "not_found" class PermissionDenied(AppError): status_code = 403 code = "forbidden" @app.exception_handler(AppError) async def app_error_handler(request: Request, exc: AppError): return JSONResponse( status_code=exc.status_code, content={"error": {"code": exc.code, "message": str(exc)}}, ) ``` Now your code can raise: ```python @app.get("/users/{user_id}") def get_user(user_id: int, db: Session = Depends(get_db)): user = db.get(User, user_id) if not user: raise NotFoundError(f"User {user_id} not found") return user ``` Cleaner than scattered HTTPException raises. ## Generic exception handler For unexpected errors: ```python @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): logger.exception( "unhandled_exception", extra={"method": request.method, "path": request.url.path}, ) return JSONResponse( status_code=500, content={"error": {"code": "internal_error", "message": "Something went wrong"}}, ) ``` Don't leak stack traces to clients. Log the full traceback; return a generic message. ## CORS middleware For browsers to call from different origins: ```python from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["https://app.example.com", "http://localhost:3000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) ``` Production: list specific origins. Avoid `allow_origins=["*"]` if you allow credentials. ## GZip middleware Compress responses: ```python from fastapi.middleware.gzip import GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) ``` Smaller responses; faster client receive. Mostly automatic. ## Custom middleware pattern ```python from starlette.middleware.base import BaseHTTPMiddleware class TimingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): start = time.perf_counter() response = await call_next(request) duration_ms = (time.perf_counter() - start) * 1000 response.headers["X-Response-Time-Ms"] = f"{duration_ms:.2f}" return response app.add_middleware(TimingMiddleware) ``` Reusable; classier than inline `@app.middleware("http")`. ## Health endpoint (