44. A production API with FastAPI: validation, limits, and shutdown

FastAPI builds an HTTP API from type-hinted functions: pydantic validates the input, the hints generate an OpenAPI schema, and it runs on an ASGI server (uvicorn) with asyncio. Even if your production services are in Go, as in this Wiki's projects, you will meet FastAPI in every ML team, usually in front of a model.

# labs/api.py   (uv add fastapi uvicorn pydantic)
import asyncio
import logging
import time
import uuid
from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator

log = logging.getLogger("api")

class TicketIn(BaseModel):
    subject: Annotated[str, Field(min_length=3, max_length=200)]
    body: Annotated[str, Field(max_length=10_000)]
    priority: Annotated[int, Field(ge=1, le=4)] = 3

    @field_validator("subject")
    @classmethod
    def no_control_chars(cls, v: str) -> str:
        if any(ord(c) < 32 for c in v):
            raise ValueError("control characters not allowed")
        return v.strip()

class TicketOut(BaseModel):
    id: str
    category: str
    priority: int

TICKETS: dict[str, TicketOut] = {}

def classify(text: str) -> str:                       # stand-in for the model from Ch 56
    t = text.lower()
    return "billing" if "refund" in t or "invoice" in t else "technical" if "error" in t else "general"

@asynccontextmanager
async def lifespan(app: FastAPI):
    log.info("startup: load model, open DB pool")     # runs once before serving
    yield
    log.info("shutdown: flush, close pools")          # runs on SIGTERM after in-flight requests finish

app = FastAPI(title="tickets", lifespan=lifespan)

@app.middleware("http")
async def request_id_and_timing(request: Request, call_next):
    rid = request.headers.get("x-request-id") or uuid.uuid4().hex
    start = time.perf_counter()
    if int(request.headers.get("content-length") or 0) > 64 * 1024:
        return JSONResponse({"detail": "body too large"}, status_code=413)
    try:
        async with asyncio.timeout(5):                # per-request deadline
            response = await call_next(request)
    except TimeoutError:
        response = JSONResponse({"detail": "timeout"}, status_code=504)
    response.headers["x-request-id"] = rid
    log.info("%s %s %s %.1fms rid=%s", request.method, request.url.path,
             response.status_code, (time.perf_counter() - start) * 1000, rid)
    return response

def api_key(request: Request) -> str:
    key = request.headers.get("x-api-key", "")
    if key != "dev-key":                              # Ch 51 replaces this with real auth
        raise HTTPException(status_code=401, detail="invalid API key")
    return key

@app.get("/healthz")
async def healthz() -> dict[str, bool]:
    return {"ok": True}

@app.post("/tickets", status_code=201)
async def create_ticket(t: TicketIn, _: Annotated[str, Depends(api_key)]) -> TicketOut:
    category = await asyncio.to_thread(classify, f"{t.subject} {t.body}")   # CPU work off the loop
    out = TicketOut(id=uuid.uuid4().hex[:12], category=category, priority=t.priority)
    TICKETS[out.id] = out
    return out

@app.get("/tickets/{ticket_id}")
async def get_ticket(ticket_id: str, _: Annotated[str, Depends(api_key)]) -> TicketOut:
    if (t := TICKETS.get(ticket_id)) is None:
        raise HTTPException(status_code=404, detail="not found")
    return t

uv run uvicorn labs.api:app --port 8000 --timeout-graceful-shutdown 20
curl -s localhost:8000/tickets -H 'x-api-key: dev-key' -H 'content-type: application/json' \
     -d '{"subject": "Refund for order 7", "body": "charged twice"}'
# {"id":"...","category":"billing","priority":3}
curl -s localhost:8000/tickets -H 'x-api-key: dev-key' -H 'content-type: application/json' \
     -d '{"subject": "x", "body": "", "priority": 9}'
# 422 with a precise list of validation errors
open http://localhost:8000/docs      # generated interactive OpenAPI docs

What makes it production-shaped, mapped to the Go server in HTTPS Ch 19:

• Validation at the edge. pydantic rejects malformed input with a 422 before your code runs. Length limits on every string.
• Body size limit and a per-request deadline in middleware. (In production, also enforce body size at the reverse proxy.)
• CPU work off the event loop (asyncio.to_thread); a model call that blocks the loop stalls every request (Ch 29.5).
• Graceful shutdown. On SIGTERM, uvicorn stops accepting, waits for in-flight requests (up to the graceful timeout), then runs the lifespan shutdown code.
• Request IDs in logs and responses, so one request can be traced across services (Ch 68).

Run several worker processes for CPU parallelism (uvicorn --workers 4, or one process per container with Kubernetes doing the replication), behind a reverse proxy that terminates TLS.