Fastapi Patterns

FastAPI Patterns

When to Use

Activate this skill when:

• Configuring FastAPI middleware (CORS, logging, timing, error handling)
• Creating complex dependency injection chains
• Implementing WebSocket endpoints with connection management
• Customizing OpenAPI documentation (tags, examples, deprecation)
• Setting up JWT authentication and role-based access dependencies
• Implementing background tasks (lightweight or distributed)
• Managing application lifecycle (startup/shutdown via lifespan)
• Setting up rate limiting or request throttling

Do NOT use this skill for:

• Basic endpoint CRUD, repository, or service patterns (use python-backend-expert)
• Writing tests for FastAPI endpoints (use pytest-patterns)
• API contract design or schema planning (use api-design-patterns)
• Architecture decisions (use system-architecture)

Instructions

Middleware Stack

Middleware executes in LIFO (Last In, First Out) order. The last middleware added is the outermost layer.

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

app = FastAPI()

# Order matters: added last = executed first (outermost)
app.add_middleware(TimingMiddleware)          # 3rd added = runs 1st
app.add_middleware(RequestLoggingMiddleware)  # 2nd added = runs 2nd
app.add_middleware(                           # 1st added = runs 3rd (innermost)
    CORSMiddleware,
    allow_origins=["https://app.example.com"],  # NEVER use "*" in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

ASGI Middleware (Preferred)

Use pure ASGI middleware for performance-critical paths:

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

class TimingMiddleware:
    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

        start = time.perf_counter()
        await self.app(scope, receive, send)
        duration = time.perf_counter() - start
        # Log or record the duration

BaseHTTPMiddleware (Simpler but Slower)

Use only for middleware that needs to read/modify the request body or response:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

When to use which:

• ASGI middleware: Performance-critical, no need to read request/response body
• BaseHTTPMiddleware: Need access to Request/Response objects, simpler API

Authentication Dependencies

JWT Token Validation

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user

Role-Based Access (Factory Pattern)

def require_role(*roles: str):
    """Factory that creates a dependency requiring specific roles."""
    async def check_role(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(
                status_code=403,
                detail=f"Requires one of: {', '.join(roles)}",
            )
        return user
    return check_role

# Usage in routes
@router.delete("/users/{user_id}", dependencies=[Depends(require_role("admin"))])
async def delete_user(user_id: int, ...) -> None:
    ...

@router.patch("/posts/{post_id}")
async def update_post(
    post_id: int,
    user: User = Depends(require_role("admin", "editor")),
) -> PostResponse:
    ...

Dependency Injection Chains

Caching Behavior

FastAPI caches dependency results within a single request. The same dependency called multiple times returns the same instance:

# get_async_session is called once per request, even if used by multiple deps
async def get_user_service(session: AsyncSession = Depends(get_async_session)) -> UserService:
    return UserService(session)

async def get_post_service(session: AsyncSession = Depends(get_async_session)) -> PostService:
    return PostService(session)  # Same session instance as user_service

To disable caching (get a new instance each time), use use_cache=False:

session: AsyncSession = Depends(get_async_session, use_cache=False)

Yield Dependencies (Resource Cleanup)

async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client
    # Client is automatically closed after the request

Overriding Dependencies in Tests

# In tests
from app.main import app
from app.dependencies.auth import get_current_user

async def mock_current_user() -> User:
    return User(id=1, email="test@example.com", role="admin")

app.dependency_overrides[get_current_user] = mock_current_user

Background Tasks

FastAPI BackgroundTasks (Lightweight)

For tasks that don't need to survive server restarts:

from fastapi import BackgroundTasks

@router.post("/users", status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    user = await service.create_user(data)
    background_tasks.add_task(send_welcome_email, user.email)
    return UserResponse.model_validate(user)

async def send_welcome_email(email: str) -> None:
    """Runs after the response is sent. Creates its own session."""
    async with async_session_factory() as session:
        async with session.begin():
            # Send email, log activity, etc.
            ...

Rules:

• Never reuse the request session in background tasks — create a new one
• Background tasks run in the same process — no retry, no persistence
• Use Celery or similar for tasks that need reliability, retry, or distribution

WebSocket Pattern

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self) -> None:
        self.active: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, ws: WebSocket) -> None:
        await ws.accept()
        self.active.setdefault(user_id, []).append(ws)

    def disconnect(self, user_id: int, ws: WebSocket) -> None:
        if user_id in self.active:
            self.active[user_id].remove(ws)
            if not self.active[user_id]:
                del self.active[user_id]

    async def send_to_user(self, user_id: int, message: dict) -> None:
        for ws in self.active.get(user_id, []):
            await ws.send_json(message)

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket, token: str) -> None:
    # Auth via query parameter: /ws?token=xxx
    user = await verify_ws_token(token)
    if not user:
        await ws.close(code=4001)
        return

    await manager.connect(user.id, ws)
    try:
        while True:
            data = await ws.receive_json()
            # Process incoming messages
            await handle_message(user.id, data)
    except WebSocketDisconnect:
        manager.disconnect(user.id, ws)

WebSocket auth approaches: 1. Query parameter: ws://host/ws?token=xxx (simplest, token visible in logs) 2. First message: Connect, then send token as first message (more secure) 3. Cookie: Use existing session cookie (requires same domain)

Application Lifespan

Use the lifespan context manager (not deprecated on_event):

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Startup: initialize resources
    await init_database()
    redis = await aioredis.from_url(settings.redis_url)
    app.state.redis = redis

    yield  # Application runs here

    # Shutdown: cleanup resources
    await redis.close()
    await dispose_engine()

app = FastAPI(lifespan=lifespan)

Lifespan responsibilities:

• Database connection pool initialization and disposal
• Redis/cache connection setup and teardown
• HTTP client pool creation
• Background scheduler startup/shutdown
• Cache warmup on startup

Exception Handlers

Register global exception handlers for consistent error responses:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "code": "VALIDATION_ERROR",
            "field_errors": [
                {"field": e["loc"][-1], "message": e["msg"], "code": e["type"]}
                for e in exc.errors()
            ],
        },
    )

OpenAPI Customization

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="API description with **markdown** support",
    openapi_tags=[
        {"name": "Users", "description": "User management operations"},
        {"name": "Auth", "description": "Authentication endpoints"},
    ],
    docs_url="/docs",        # Swagger UI
    redoc_url="/redoc",      # ReDoc
    openapi_url="/openapi.json",
)

Examples

JWT Auth Dependency Chain

Complete auth chain from token to authorized user:

Request with Authorization: Bearer <token>
    ↓
oauth2_scheme (extracts token from header)
    ↓
get_current_user (decodes JWT, loads user from DB)
    ↓
require_role("admin") (checks user.role)
    ↓
Route handler (receives verified admin user)

Each dependency in the chain is independently testable via dependency_overrides.

Edge Cases

• Middleware vs dependency: Use middleware for cross-cutting concerns (logging, timing, CORS). Use dependencies for per-route logic (auth, pagination params, feature flags).

• ASGI vs BaseHTTPMiddleware: Prefer ASGI middleware for performance. BaseHTTPMiddleware reads the entire response body into memory, causing issues with streaming and large responses.

• Lifespan vs on_event: Always use the lifespan context manager. @app.on_event("startup") and @app.on_event("shutdown") are deprecated in FastAPI 0.109+.

• Depends caching across sub-applications: Dependency caching works per-request within a single app instance. If using app.mount() for sub-applications, each sub-app has its own dependency resolution scope.

• WebSocket scaling: A single server instance holds all WebSocket connections. For multi-instance deployments, use Redis pub/sub to broadcast messages across instances.

See references/middleware-examples.md for complete middleware implementations. See references/dependency-injection-patterns.md for advanced DI patterns.

Dependency Injection Patterns

Advanced FastAPI dependency injection patterns for authentication, authorization, resource management, and testing.

---

Pattern 1: Authentication Chain

Build a chain from token extraction to authorized user:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.database import get_async_session
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    """Decode JWT and load user from database."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id: int | None = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise credentials_exception
    return user


async def get_current_active_user(
    user: User = Depends(get_current_user),
) -> User:
    """Verify user account is active."""
    if not user.is_active:
        raise HTTPException(status_code=403, detail="Account deactivated")
    return user

---

Pattern 2: Role-Based Access Factory

Create parameterized dependencies with a factory function:

def require_role(*allowed_roles: str):
    """Factory: creates a dependency that checks user roles."""
    async def _check_role(
        user: User = Depends(get_current_active_user),
    ) -> User:
        if user.role not in allowed_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Required role: {' or '.join(allowed_roles)}",
            )
        return user
    return _check_role


# Usage
@router.get("/admin/dashboard")
async def admin_dashboard(
    admin: User = Depends(require_role("admin")),
) -> dict:
    ...

@router.put("/posts/{post_id}")
async def update_post(
    post_id: int,
    user: User = Depends(require_role("admin", "editor")),
) -> PostResponse:
    ...

---

Pattern 3: Resource Ownership Check

Verify the authenticated user owns the resource:

def require_ownership(resource_type: str):
    """Factory: checks that the current user owns the resource."""
    async def _check_ownership(
        resource_id: int,
        user: User = Depends(get_current_active_user),
        session: AsyncSession = Depends(get_async_session),
    ) -> User:
        # Dynamic lookup based on resource type
        model_map = {"post": Post, "comment": Comment}
        model = model_map.get(resource_type)
        if model is None:
            raise ValueError(f"Unknown resource type: {resource_type}")

        resource = await session.get(model, resource_id)
        if resource is None:
            raise HTTPException(status_code=404, detail=f"{resource_type} not found")
        if resource.user_id != user.id and user.role != "admin":
            raise HTTPException(status_code=403, detail="Not the owner")
        return user
    return _check_ownership


@router.delete("/posts/{resource_id}")
async def delete_post(
    resource_id: int,
    user: User = Depends(require_ownership("post")),
) -> None:
    ...

---

Pattern 4: Pagination Parameters

Reusable pagination dependency:

from pydantic import BaseModel, Field


class PaginationParams(BaseModel):
    cursor: str | None = None
    limit: int = Field(default=20, ge=1, le=100)


async def get_pagination(
    cursor: str | None = None,
    limit: int = 20,
) -> PaginationParams:
    return PaginationParams(cursor=cursor, limit=min(max(limit, 1), 100))


@router.get("/posts")
async def list_posts(
    pagination: PaginationParams = Depends(get_pagination),
) -> PostListResponse:
    ...

---

Pattern 5: Yield Dependencies (Resource Lifecycle)

Use yield for dependencies that need cleanup:

import httpx
from collections.abc import AsyncGenerator


async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    """HTTP client with automatic cleanup."""
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        timeout=30.0,
        headers={"Accept": "application/json"},
    ) as client:
        yield client
    # Client is closed after response is sent


@router.get("/external-data")
async def fetch_external(
    client: httpx.AsyncClient = Depends(get_http_client),
) -> dict:
    response = await client.get("/data")
    return response.json()

---

Pattern 6: Cached Dependencies

FastAPI caches dependency results per request by default:

# Both services get the SAME session instance (cached per request)
@router.post("/transfer")
async def transfer(
    user_service: UserService = Depends(get_user_service),    # Uses session A
    order_service: OrderService = Depends(get_order_service),  # Uses session A (same!)
) -> TransferResponse:
    # Both services share the same transaction
    ...

To get separate instances, disable caching:

async def get_random_id() -> str:
    return str(uuid.uuid4())

@router.get("/test")
async def test(
    id1: str = Depends(get_random_id),                       # Same value
    id2: str = Depends(get_random_id),                       # Same value (cached!)
    id3: str = Depends(get_random_id, use_cache=False),      # Different value
) -> dict:
    return {"id1": id1, "id2": id2, "id3": id3}

---

Pattern 7: Overriding Dependencies in Tests

import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app
from app.dependencies.auth import get_current_user
from app.models.user import User


@pytest.fixture
def mock_admin_user():
    return User(id=1, email="admin@test.com", role="admin", is_active=True)


@pytest.fixture
async def client(mock_admin_user):
    """Test client with mocked auth."""

    async def override_current_user():
        return mock_admin_user

    app.dependency_overrides[get_current_user] = override_current_user

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()

---

Pattern 8: API Key Authentication

Alternative to JWT for service-to-service communication:

from fastapi import Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")


async def verify_api_key(
    api_key: str = Security(api_key_header),
) -> str:
    if api_key not in settings.valid_api_keys:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key


@router.get("/internal/metrics", dependencies=[Depends(verify_api_key)])
async def get_metrics() -> dict:
    ...

---

Dependency Resolution Order

Route handler parameters resolved left-to-right:
  @router.get("/")
  async def handler(
      a: A = Depends(get_a),     # 1st: resolved
      b: B = Depends(get_b),     # 2nd: resolved
      c: C = Depends(get_c),     # 3rd: resolved
  ):

Nested dependencies resolved depth-first:
  get_a depends on get_x, get_y
  Resolution: get_x → get_y → get_a → get_b → get_c

Shared dependencies resolved once (cached):
  If get_a and get_b both depend on get_session,
  get_session is called once and the result is shared.

Middleware Examples

Complete middleware implementations for FastAPI applications.

---

1. Request Logging Middleware

Logs every request with method, path, status code, and duration.

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

logger = logging.getLogger("api.access")


class RequestLoggingMiddleware:
    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

        start = time.perf_counter()
        status_code = 500  # Default if response fails

        async def send_wrapper(message):
            nonlocal status_code
            if message["type"] == "http.response.start":
                status_code = message["status"]
            await send(message)

        try:
            await self.app(scope, receive, send_wrapper)
        finally:
            duration = time.perf_counter() - start
            method = scope.get("method", "?")
            path = scope.get("path", "?")
            logger.info(
                "request completed",
                extra={
                    "method": method,
                    "path": path,
                    "status_code": status_code,
                    "duration_ms": round(duration * 1000, 2),
                },
            )

---

2. Request ID Middleware

Assigns a unique ID to each request for tracing.

import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response


class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id

        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

---

3. Rate Limiting Middleware

Token bucket rate limiter with per-IP tracking.

import time
from collections import defaultdict
from starlette.types import ASGIApp, Receive, Scope, Send
from starlette.responses import JSONResponse


class RateLimitMiddleware:
    def __init__(
        self,
        app: ASGIApp,
        *,
        requests_per_minute: int = 60,
    ) -> None:
        self.app = app
        self.rate = requests_per_minute
        self.window = 60.0
        self.buckets: dict[str, list[float]] = defaultdict(list)

    def _get_client_ip(self, scope: Scope) -> str:
        # Check for proxy headers
        headers = dict(scope.get("headers", []))
        forwarded = headers.get(b"x-forwarded-for", b"").decode()
        if forwarded:
            return forwarded.split(",")[0].strip()
        client = scope.get("client")
        return client[0] if client else "unknown"

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

        client_ip = self._get_client_ip(scope)
        now = time.time()

        # Clean old entries
        self.buckets[client_ip] = [
            t for t in self.buckets[client_ip] if now - t < self.window
        ]

        if len(self.buckets[client_ip]) >= self.rate:
            response = JSONResponse(
                status_code=429,
                content={"detail": "Rate limit exceeded", "code": "RATE_LIMITED"},
                headers={
                    "Retry-After": str(int(self.window)),
                    "X-RateLimit-Limit": str(self.rate),
                    "X-RateLimit-Remaining": "0",
                },
            )
            await response(scope, receive, send)
            return

        self.buckets[client_ip].append(now)
        await self.app(scope, receive, send)

---

4. Error Handling Middleware

Catches unhandled exceptions and returns consistent error responses.

import logging
import traceback
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse

logger = logging.getLogger("api.errors")


class ErrorHandlingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:
            logger.error(
                "Unhandled exception",
                extra={
                    "path": request.url.path,
                    "method": request.method,
                    "error": str(exc),
                    "traceback": traceback.format_exc(),
                },
            )
            return JSONResponse(
                status_code=500,
                content={
                    "detail": "Internal server error",
                    "code": "INTERNAL_ERROR",
                },
            )

---

5. CORS Configuration Examples

Development (Permissive)

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Production (Restrictive)

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://app.example.com",
        "https://admin.example.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
    expose_headers=["X-Request-ID", "X-RateLimit-Remaining"],
    max_age=600,  # Cache preflight for 10 minutes
)

---

Recommended Middleware Order

Add middleware in this order (last added = outermost = executes first):

# 5. Error handling (outermost — catches everything)
app.add_middleware(ErrorHandlingMiddleware)

# 4. Request timing/logging
app.add_middleware(RequestLoggingMiddleware)

# 3. Request ID assignment
app.add_middleware(RequestIdMiddleware)

# 2. Rate limiting
app.add_middleware(RateLimitMiddleware, requests_per_minute=100)

# 1. CORS (innermost — closest to routes)
app.add_middleware(CORSMiddleware, ...)

This ensures:

• CORS headers are always included (even on errors)
• Rate limiting runs before route processing
• Every request gets an ID before logging
• All requests are logged with timing
• Unhandled exceptions are caught and formatted