Middleware & CORS — Cheat Sheet

Middleware — Code Around Every Request

import logging, time, uuid
from fastapi import FastAPI, Request

app = FastAPI()
log = logging.getLogger("access")

@app.middleware("http")
async def observability(request: Request, call_next):
    # ── inbound: before the route (and its dependencies) run ──
    request_id = request.headers.get("x-request-id", uuid.uuid4().hex[:12])
    request.state.request_id = request_id        # visible to endpoints & handlers
    start = time.perf_counter()

    response = await call_next(request)          # ← the entire app runs here

    # ── outbound: response exists, decorate + log ──
    elapsed_ms = (time.perf_counter() - start) * 1000
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Response-Time"] = f"{elapsed_ms:.1f}ms"
    log.info("%s %s → %d in %.1fms id=%s",
             request.method, request.url.path,
             response.status_code, elapsed_ms, request_id)
    return response

@app.get("/orders/{order_id}")
def get_order(order_id: int, request: Request):
    # the id set by middleware, usable in logs / error reports:
    return {"order": order_id, "trace": request.state.request_id}

# access log:  GET /orders/42 → 200 in 3.2ms id=7f3a99c1e0b2
# The gateway sets X-Request-ID once; every service propagates it →
# one grep traces a user's request across the whole system.

CORS — Letting Browsers Call Your API

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=[                       # EXACT matches — no trailing slash
        "http://localhost:3000",          # Next.js dev
        "https://aicancode.org",          # production frontend
    ],
    allow_credentials=True,               # cookies / Authorization allowed
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
    max_age=600,                          # cache preflight verdict 10 min
)

@app.get("/api/courses")
def courses():
    return [{"id": 1, "title": "Python A-Z"}]

# What actually happens for fetch("http://localhost:8000/api/courses",
#                                {headers: {Authorization: "Bearer ..."}}):
#
# 1. Browser preflights (because of the Authorization header):
#    OPTIONS /api/courses
#    Origin: http://localhost:3000
#    Access-Control-Request-Method: GET
#    Access-Control-Request-Headers: authorization
# 2. Middleware answers (no route code runs):
#    Access-Control-Allow-Origin: http://localhost:3000
#    Access-Control-Allow-Methods: GET, POST, PUT, PATCH, DELETE
#    Access-Control-Allow-Headers: Authorization, Content-Type, X-Request-ID
#    Access-Control-Allow-Credentials: true
# 3. Browser proceeds with the real GET; response carries
#    Access-Control-Allow-Origin again; your JS can now read it.
# Without step 2's headers → the request may even EXECUTE,
# but the browser refuses to let your JavaScript see the response.

Lifespan — Startup & Shutdown Done Properly

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── STARTUP: before the first request is accepted ──
    app.state.http = httpx.AsyncClient(timeout=10)      # reused pool, all requests
    app.state.model = load_recommender("model-v3.bin")  # 2 GB of weights, ONCE
    # engine/redis init also lives here in real apps
    print("startup complete — accepting traffic")

    yield                                               # ← the app serves here

    # ── SHUTDOWN: SIGTERM received, in-flight requests draining ──
    await app.state.http.aclose()                       # return connections politely
    print("shutdown clean")

app = FastAPI(lifespan=lifespan)

@app.get("/recommendations/{student_id}")
async def recommend(student_id: int, request: Request):
    model = request.app.state.model                     # shared, already loaded
    jobs = model.recommend(student_id, k=5)
    # shared client → connection reuse instead of a new TCP+TLS handshake:
    r = await request.app.state.http.get("https://api.internal/enrich")
    return {"student": student_id, "jobs": jobs, "enriched": r.status_code == 200}

# Legacy spelling you'll still see (deprecated — migrate on sight):
#   @app.on_event("startup")
#   async def start(): ...
# One lifespan function keeps setup and teardown adjacent and testable —
# TestClient runs the full lifespan too: with TestClient(app) as client: ...