Production Fastapi

Production FastAPI

This skill encodes battle-tested patterns for shipping FastAPI applications that survive real production traffic. Every recommendation here comes from outage post-mortems, not blog posts. The patterns are opinionated — this is a senior engineer review, not a tutorial.

See templates/ for copy-paste-ready implementations of every pattern below.

---

1. Application Lifecycle

Use lifespan, not on_event. The on_event decorator is deprecated and does not support shared state between startup and shutdown.

from contextlib import asynccontextmanager
from fastapi import FastAPI
import httpx
from sqlalchemy.ext.asyncio import create_async_engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- Startup ---
    app.state.db_engine = create_async_engine(
        settings.database_url,
        pool_size=settings.db_pool_size,
        max_overflow=settings.db_max_overflow,
        pool_pre_ping=True,
        pool_recycle=3600,
    )
    app.state.http_client = httpx.AsyncClient(
        timeout=httpx.Timeout(10.0, connect=5.0),
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    )
    configure_logging(settings.environment)
    logger.info("application_started", version=settings.app_version)

    yield

    # --- Shutdown ---
    await app.state.http_client.aclose()
    await app.state.db_engine.dispose()
    logger.info("application_stopped")

app = FastAPI(title="My Service", lifespan=lifespan)

Why lifespan matters:

• Resources created in startup are guaranteed to be cleaned up in shutdown
• The yield pattern makes it impossible to forget cleanup
• app.state shares resources across the request lifecycle without globals
• If startup fails, the app never starts accepting traffic

Startup checklist:

1. Database engine/connection pool
2. HTTP client pool (reuse connections)
3. Logging configuration
4. Cache connections (Redis)
5. Background task schedulers

Shutdown checklist (reverse order):

1. Cancel background tasks
2. Close HTTP client pools
3. Dispose database engine (waits for active connections)
4. Flush log buffers

---

2. Structured Logging

Use structlog with JSON output in production, pretty console in development. Never use print() or logging.basicConfig().

import structlog
import logging
from contextvars import ContextVar

request_id_ctx: ContextVar[str | None] = ContextVar("request_id", default=None)

def configure_logging(environment: str) -> None:
    shared_processors: list[structlog.types.Processor] = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.format_exc_info,
    ]

    if environment == "production":
        renderer = structlog.processors.JSONRenderer()
    else:
        renderer = structlog.dev.ConsoleRenderer()

    structlog.configure(
        processors=[
            *shared_processors,
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            renderer,
        ],
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.handlers.clear()
    root_logger.addHandler(handler)
    root_logger.setLevel(logging.INFO)

    # Silence noisy libraries
    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
    logging.getLogger("httpx").setLevel(logging.WARNING)

CRITICAL: Never use BaseHTTPMiddleware for logging or request ID injection. It breaks contextvars because it runs the endpoint in a different task than the middleware. Use pure ASGI middleware instead:

import uuid
import structlog
from starlette.types import ASGIApp, Receive, Scope, Send

class RequestIDMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] not in ("http", "websocket"):
            await self.app(scope, receive, send)
            return

        request_id = str(uuid.uuid4())
        scope.setdefault("state", {})["request_id"] = request_id
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(request_id=request_id)

        async def send_with_request_id(message):
            if message["type"] == "http.response.start":
                headers = list(message.get("headers", []))
                headers.append((b"x-request-id", request_id.encode()))
                message["headers"] = headers
            await send(message)

        await self.app(scope, receive, send_with_request_id)

See templates/logging_config.py for the complete structlog setup and templates/middleware.py for all ASGI middleware implementations.

---

3. Health Checks

Every production service needs two health endpoints. No exceptions.

from fastapi import APIRouter, Response
from datetime import datetime, UTC
import asyncio

health_router = APIRouter(tags=["health"])

@health_router.get("/health/live")
async def liveness() -> dict:
    """Is the process alive? Always 200. Used by load balancers."""
    return {"status": "alive", "timestamp": datetime.now(UTC).isoformat()}

@health_router.get("/health/ready")
async def readiness(request: Request) -> Response:
    """Can we handle requests? Checks all dependencies."""
    checks = {}
    all_healthy = True

    # Database check with timeout
    try:
        async with asyncio.timeout(2.0):
            async with request.app.state.db_engine.connect() as conn:
                await conn.execute(text("SELECT 1"))
        checks["database"] = {"status": "healthy", "latency_ms": ...}
    except Exception as e:
        checks["database"] = {"status": "unhealthy", "error": str(e)}
        all_healthy = False

    # Redis check with timeout
    try:
        async with asyncio.timeout(2.0):
            await request.app.state.redis.ping()
        checks["redis"] = {"status": "healthy"}
    except Exception as e:
        checks["redis"] = {"status": "unhealthy", "error": str(e)}
        all_healthy = False

    status_code = 200 if all_healthy else 503
    return JSONResponse(
        status_code=status_code,
        content={
            "status": "ready" if all_healthy else "not_ready",
            "checks": checks,
            "timestamp": datetime.now(UTC).isoformat(),
        },
    )

Health check rules:

• /health/live — NEVER check dependencies. If the process is running, return 200. Used by load balancers to detect crashed processes.
• /health/ready — Check every dependency with a 2-second timeout each. Used by Kubernetes readiness probes and orchestrators.
• Exclude health endpoints from access logs (noise).
• Exclude health endpoints from authentication middleware.
• Return structured JSON, not just a status code — operators need to see which dependency is down.

See templates/health_checks.py for the complete implementation with timing, startup checks, and Kubernetes probe configuration.

---

4. Error Handling

Use RFC 7807 Problem Details format. Consistent error responses are non-negotiable in production.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import structlog

logger = structlog.get_logger()

class AppError(Exception):
    def __init__(
        self,
        detail: str,
        status_code: int = 500,
        error_type: str = "internal_error",
    ):
        self.detail = detail
        self.status_code = status_code
        self.error_type = error_type

async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    request_id = request.state.request_id if hasattr(request.state, "request_id") else None
    logger.warning("handled_error", error_type=exc.error_type, detail=exc.detail)
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "type": f"https://api.example.com/errors/{exc.error_type}",
            "title": exc.error_type.replace("_", " ").title(),
            "status": exc.status_code,
            "detail": exc.detail,
            "instance": str(request.url),
            "request_id": request_id,
        },
    )

async def unhandled_error_handler(request: Request, exc: Exception) -> JSONResponse:
    request_id = request.state.request_id if hasattr(request.state, "request_id") else None
    logger.exception("unhandled_error", request_id=request_id)

    # Sentry capture (if configured)
    try:
        import sentry_sdk
        sentry_sdk.capture_exception(exc)
    except ImportError:
        pass

    # NEVER expose tracebacks in production
    return JSONResponse(
        status_code=500,
        content={
            "type": "https://api.example.com/errors/internal_error",
            "title": "Internal Server Error",
            "status": 500,
            "detail": "An unexpected error occurred. Please contact support.",
            "request_id": request_id,
        },
    )

app.add_exception_handler(AppError, app_error_handler)
app.add_exception_handler(Exception, unhandled_error_handler)

Error handling rules:

• Every error response includes request_id — this is the lifeline for debugging in production
• Known errors (validation, auth, not found) return specific status codes and human-readable messages
• Unknown errors return 500 with a generic message — never expose tracebacks, SQL queries, or internal paths
• Log every error with full context (the log has the traceback; the response does not)
• Send unexpected errors to Sentry/error tracking with the request_id

See templates/error_handlers.py for the complete exception hierarchy and handler registration.

---

5. Middleware Stack

Order matters. Middleware executes in an onion model — first added is outermost. Design accordingly:

# Add in this order (outermost to innermost):
app.add_middleware(CORSMiddleware, ...)          # 1. CORS (must run first)
app.add_middleware(SecurityHeadersMiddleware)     # 2. Security headers
app.add_middleware(RequestIDMiddleware)           # 3. Request ID generation
app.add_middleware(LoggingMiddleware)             # 4. Request/response logging
# Auth and rate limiting typically via Depends(), not middleware

CORS: Never Wildcard With Credentials

from fastapi.middleware.cors import CORSMiddleware

# GOOD — specific origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com", "https://staging.example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

# DANGEROUS — wildcard + credentials is a security vulnerability
# Browsers ignore Access-Control-Allow-Credentials when origin is *,
# but this signals you haven't thought about CORS policy.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],         # NEVER with allow_credentials=True
    allow_credentials=True,      # This combination is a red flag
)

Pure ASGI Middleware Template

Always use this pattern instead of BaseHTTPMiddleware:

from starlette.types import ASGIApp, Receive, Scope, Send

class MyMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        # --- Pre-processing ---
        # Modify scope, set contextvars, start timer, etc.

        async def send_wrapper(message):
            if message["type"] == "http.response.start":
                # --- Modify response headers ---
                pass
            await send(message)

        await self.app(scope, receive, send_wrapper)
        # --- Post-processing ---

Why not BaseHTTPMiddleware?

• It reads the entire request body into memory (no streaming)
• It runs the endpoint in a separate anyio task, breaking contextvars
• It catches exceptions, making error handling middleware unreliable
• Pure ASGI middleware has zero overhead and full control

See templates/middleware.py for RequestID, Logging, and SecurityHeaders implementations.

---

6. Pydantic v2

Pydantic v2 has breaking API changes. Use the new API everywhere.

from pydantic import BaseModel, Field, model_validator, ConfigDict
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Annotated
from datetime import datetime, UTC

# --- Models: Use Annotated + Field ---
class CreateUserRequest(BaseModel):
    model_config = ConfigDict(strict=True)  # Strict mode: no coercion

    email: Annotated[str, Field(max_length=255, pattern=r"^[\w.-]+@[\w.-]+\.\w+$")]
    name: Annotated[str, Field(min_length=1, max_length=100)]
    age: Annotated[int, Field(ge=0, le=150)] | None = None

    @model_validator(mode="after")
    def validate_model(self):
        # Cross-field validation goes here
        return self

# --- Response models: Never expose internal fields ---
class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # ORM mode

    public_id: str
    email: str
    name: str
    created_at: datetime

# --- Serialization: model_dump() not .dict() ---
user = UserResponse.model_validate(db_user)  # NOT parse_obj()
data = user.model_dump(exclude_none=True)     # NOT .dict()
json_str = user.model_dump_json()             # NOT .json()

# --- Settings: Use SettingsConfigDict ---
class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

    app_name: str = "my-service"
    app_version: str = "0.1.0"
    environment: str = "development"
    debug: bool = False

    database_url: str
    db_pool_size: int = 10
    db_max_overflow: int = 5

    redis_url: str = "redis://localhost:6379/0"
    cors_origins: list[str] = ["http://localhost:3000"]
    log_level: str = "INFO"

Pydantic v2 migration cheat sheet:

| v1 (deprecated) | v2 (use this) | |-----------------|---------------| | .dict() | .model_dump() | | .json() | .model_dump_json() | | .parse_obj() | .model_validate() | | .parse_raw() | .model_validate_json() | | class Config: | model_config = ConfigDict(...) | | @validator | @field_validator | | @root_validator | @model_validator | | orm_mode = True | from_attributes = True | | Field(regex=...) | Field(pattern=...) |

Rules:

• Use strict=True on models that accept external input — prevents type coercion attacks
• Use from_attributes=True on response models that serialize from ORM objects
• Settings must fail fast — if database_url is missing, the app crashes at import time, not on first request
• Never expose internal model fields (like id, password_hash) in response models

---

7. Async Patterns

The cardinal sin of async FastAPI: blocking the event loop with synchronous I/O.

# WRONG — blocks the event loop, starves all other requests
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    response = requests.get(f"https://api.example.com/users/{user_id}")  # BLOCKS
    data = open("config.json").read()  # BLOCKS
    time.sleep(1)  # BLOCKS
    return response.json()

# RIGHT — all I/O is async with explicit timeouts
@app.get("/users/{user_id}")
async def get_user(user_id: int, request: Request):
    async with asyncio.timeout(5.0

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [vstorm-co](https://github.com/vstorm-co)
- **Source:** [vstorm-co/production-stack-skills](https://github.com/vstorm-co/production-stack-skills)
- **License:** MIT
- **Homepage:** https://vstorm.co/

Install and usage instructions live in the source repository linked above.