Lesson 3: FastAPI in Production

Routers, dependencies, and request models

# src/recipe_extractor/api/routes/extraction.py
from fastapi import APIRouter, Depends, status

router = APIRouter(prefix="/v1/extractions", tags=["extraction"])


class ExtractionRequest(BaseModel):
    document_id: str = Field(min_length=1, max_length=64)
    schema_version: Literal["v1", "v2"] = "v2"


class ExtractionResponse(BaseModel):
    document_id: str
    title: str
    ingredients: list[IngredientResponse]
    prompt_version: str
    cost_usd: str


@router.post("", status_code=status.HTTP_200_OK)
async def create_extraction(
    body: ExtractionRequest,
    service: ExtractionService = Depends(get_extraction_service),
    user: AuthenticatedUser = Depends(require_user),
) -> ExtractionResponse:
    """Extract structured data from a previously ingested document."""
    async with request_scope(feature="extraction", user_id=user.id):
        result = await service.extract(body.document_id)
    return to_response(result)

Separate API models from domain records. ExtractionResponse is not ExtractedRecipe. They look similar today and diverge tomorrow, and coupling them means a domain change becomes a breaking API change. The to_response function is the boundary, and it is also where you decide what not to expose.

Version the URL from the first release. Adding /v1 later requires changing every client.

Dependencies come from the composition root.

def get_extraction_service(request: Request) -> ExtractionService:
    """Resolve the service built at startup, rather than constructing one here."""
    return request.app.state.deps.extraction_service

Building a service per request creates connection pools per request, which is the Module 7 mistake in a new place.

Middleware and authentication

@app.middleware("http")
async def add_request_context(request: Request, call_next):
    """Establish a correlation identifier and record the request outcome."""
    rid = request.headers.get("x-request-id") or str(uuid.uuid4())
    token = request_id.set(rid)
    started = time.monotonic()
    try:
        response = await call_next(request)
    finally:
        request_id.reset(token)
    response.headers["x-request-id"] = rid
    logger.info(
        "http_request",
        extra={"method": request.method, "path": request.url.path,
               "status": response.status_code, "duration_ms": _ms(started)},
    )
    return response

Accepting an inbound x-request-id is what lets a trace span more than one service, and returning it lets a user quote it in a support request.

Authentication as a dependency, not middleware, because dependencies can be applied per route and appear in the generated schema.

async def require_user(
    credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer()),
    deps: Dependencies = Depends(get_deps),
) -> AuthenticatedUser:
    user = await deps.auth.verify(credentials.credentials)
    if user is None:
        raise HTTPException(status_code=401, detail="invalid credentials")
    return user

One caution about middleware in async applications. Middleware wraps every request, so anything slow or blocking in it affects everything, which is the Module 8 trap at its most damaging. Keep middleware to context, logging, and cheap header work.

Error boundaries

Domain exceptions must become HTTP responses in exactly one place.

# src/recipe_extractor/api/errors.py

STATUS_MAP: dict[type[AppError], int] = {
    ValidationError: 422,
    RetrievalError: 404,
    QuotaExceededError: 429,
    BudgetExceededError: 503,
    RateLimitError: 429,
    ProviderTimeoutError: 504,
    ProviderOverloadedError: 503,
    ProviderError: 502,
    OutputError: 502,
}


def install_error_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
        status_code = next(
            (code for cls, code in STATUS_MAP.items() if isinstance(exc, cls)), 500
        )
        logger.warning(
            "request_failed",
            extra={"error_type": type(exc).__name__, "status": status_code},
        )
        return JSONResponse(
            status_code=status_code,
            content={"error": type(exc).__name__, "message": str(exc),
                     "request_id": request_id.get()},
            headers=_retry_headers(exc),
        )

    @app.exception_handler(Exception)
    async def handle_unexpected(request: Request, exc: Exception) -> JSONResponse:
        logger.exception("unhandled_exception")
        return JSONResponse(
            status_code=500,
            content={"error": "internal_error",
                     "message": "An unexpected error occurred.",
                     "request_id": request_id.get()},
        )

Four decisions here.

The Module 5 hierarchy pays off directly. Because errors were designed as a domain hierarchy, one mapping covers everything, and route handlers contain no error translation at all.

Expected errors are warnings, unexpected ones are exceptions with a stack trace. Module 11 noted that treating recoveries as errors makes your error rate meaningless, and the same applies to a 404.

The catch-all returns no internal detail and logs everything. An exception message can contain a file path, a query fragment, or part of a prompt.

Every error response carries the request identifier, which is what turns a user report into a log query.

_retry_headers adds Retry-After for 429 and 503, closing the loop with Module 5: your service should tell its clients what your providers tell you.

Health checks and readiness probes

Two endpoints answering two different questions, and conflating them causes outages.

@router.get("/healthz", include_in_schema=False)
async def liveness() -> dict[str, str]:
    """Is the process alive? No dependency checks. Never fails on a dependency."""
    return {"status": "ok"}


@router.get("/readyz", include_in_schema=False)
async def readiness(deps: Dependencies = Depends(get_deps)) -> JSONResponse:
    """Can this instance serve traffic right now?"""
    checks = {
        "database": await _safe_check(deps.repository.ping),
        "cache": await _safe_check(deps.cache.ping),
        "retriever": await _safe_check(deps.retriever.ping),
        "provider_circuit": deps.provider_breaker.state != "open",
    }
    ready = all(checks.values())
    return JSONResponse(
        status_code=200 if ready else 503,
        content={"ready": ready, "checks": checks},
    )

Liveness must not check dependencies. A liveness probe failing causes a restart, and restarting your service because the database is briefly unavailable turns a recoverable dependency blip into a restart storm across every instance.

Readiness may check dependencies, because failing it removes the instance from the load balancer without killing it, and it recovers when the dependency does.

Include the circuit breaker state. Module 5's breaker knows the provider is down before any request does, and reporting it as not ready lets traffic route to instances whose breakers are closed.

Keep both fast and cheap. These run every few seconds on every instance, so a readiness check making a model call is a real cost, and one without a timeout can hang the probe.

Graceful shutdown with in-flight streams

Deployment terminates instances constantly. Doing it badly drops requests on every release.

@app.on_event is deprecated. The lifespan context manager is the correct pattern. [VOLATILE: confirm the current status before publishing.]

from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Build dependencies at startup, release them at shutdown."""
    settings = Settings()
    configure_logging(level=settings.log_level, json_output=settings.environment != "local")

    deps = await build_dependencies(settings)
    await verify_startup(settings, deps)
    app.state.deps = deps
    app.state.shutting_down = False

    logger.info("startup_complete", extra={"environment": settings.environment})
    try:
        yield
    finally:
        app.state.shutting_down = True
        logger.info("shutdown_started")

        await asyncio.sleep(settings.drain_delay_seconds)     # let load balancers notice

        try:
            async with asyncio.timeout(settings.shutdown_timeout_seconds):
                await deps.wait_for_inflight()
        except TimeoutError:
            logger.warning("shutdown_timeout", extra={"inflight": deps.inflight_count()})

        await deps.aclose()
        logger.info("shutdown_complete")


app = FastAPI(lifespan=lifespan)

The shutdown sequence, in order and each step for a reason.

Mark the instance as shutting down so readiness starts failing. Wait a few seconds, because load balancers poll readiness and removing yourself from rotation is not instantaneous; skipping this delay is the most common cause of dropped requests during deploys. Stop accepting new work while finishing what is in flight. Wait for in-flight requests with a bounded timeout, since an unbounded wait means a stuck request blocks the deployment. Then close connection pools, flush logs, and exit.

In-flight streams need explicit handling. A streaming response from Module 8 can last minutes, and terminating one mid-flight leaves the client with a truncated answer and no error, which is the failure that module warned about.

async def wait_for_inflight(self) -> None:
    """Wait for streams to finish, sending a shutdown event if they run long."""
    while self._active_streams:
        if time.monotonic() - self._shutdown_started > self._stream_grace_seconds:
            for stream in list(self._active_streams):
                await stream.send_event("error", {"error": "server_shutting_down",
                                                  "partial": True})
                stream.close()
            return
        await asyncio.sleep(0.5)

Telling the client that the output is partial is Module 8's mid-stream error channel and Module 5's honest degradation, applied at shutdown.

[IMAGE PROMPT M12-2
Purpose: Show the ordered graceful shutdown sequence and what dropping each step costs.
Visual type: Ordered sequence diagram with failure annotations.
Prompt: A clean educational diagram showing six numbered steps arranged left to right along a horizontal timeline, each drawn as a labelled box. Step 1 "SIGTERM received". Step 2 "mark shutting down, readiness starts failing". Step 3 "drain delay, load balancer removes instance". Step 4 "stop accepting new requests". Step 5 "wait for in-flight work, bounded timeout" with a small inner note reading "streams get a partial error event if they exceed the grace period". Step 6 "close pools, flush logs, exit". Beneath steps 2, 3, and 5, three downward callouts describe what happens if that step is skipped, reading respectively "skipped: traffic keeps arriving", "skipped: requests dropped mid-flight, the most common deploy failure", and "skipped: connections leak and streams truncate silently". A bracket above steps 3 to 5 is labelled "bounded, or a stuck request blocks the deployment".
Required elements: Six numbered ordered steps with their labels, the inner note on step 5, three skip-consequence callouts under the correct steps, a bounding bracket above the middle steps.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector.
Layout: Horizontal sequence reading left to right, consequence callouts beneath, bracket above.
Text labels: "SIGTERM received", "mark shutting down, readiness starts failing", "drain delay, load balancer removes instance", "stop accepting new requests", "wait for in-flight work, bounded timeout", "streams get a partial error event if they exceed the grace period", "close pools, flush logs, exit", "skipped: traffic keeps arriving", "skipped: requests dropped mid-flight, the most common deploy failure", "skipped: connections leak and streams truncate silently", "bounded, or a stuck request blocks the deployment".
Aspect ratio: 16:9
Accessibility: Number every step and state each consequence in text rather than relying on position or colour.
Avoid: Kubernetes or vendor logos, decorative icons, tiny text, watermarks.
Alt text: Six-step graceful shutdown sequence from SIGTERM through marking readiness failed, a drain delay, refusing new requests, bounded waiting for in-flight work including streams, and closing pools, with annotations showing what breaks when each step is skipped.
END IMAGE PROMPT]