fastapi

FastAPI Patterns

Modern FastAPI (0.100+) with Pydantic v2, async-first, and typed throughout.

When to Activate

• Structuring a FastAPI app with routers and layered architecture
• Designing dependency injection chains with Depends
• Defining Pydantic v2 request/response schemas
• Handling errors, custom exception handlers, or middleware
• Adding authentication (OAuth2, JWT, API keys)
• Writing background tasks or startup/shutdown logic
• Testing FastAPI routes with TestClient or async httpx

---

Project Structure

src/
├── api/
│   ├── app.py              # create_app(), register routers + middleware
│   ├── dependencies.py     # shared Depends (db session, current user, etc.)
│   ├── middleware.py        # CORS, logging, request ID
│   └── routes/
│       ├── users.py
│       └── orders.py
├── domain/
│   ├── entities/           # Pure Pydantic models — no ORM, no HTTP
│   ├── use_cases/          # Business logic — orchestrates services
│   └── repositories/       # Abstract interfaces (Protocol or ABC)
├── adapters/
│   ├── database/           # SQLAlchemy models + session factory
│   ├── crud/               # Concrete repository implementations
│   └── external/           # Third-party HTTP clients
├── config/
│   ├── settings.py         # Pydantic Settings (env vars)
│   └── dependencies.py     # App-wide singletons (DB engine, Redis, etc.)
└── main.py                 # uvicorn entry point

---

App Factory

# api/app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from config.dependencies import GlobalDependencies
from api.routes import users, orders


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize singletons, DB pools, caches
    await GlobalDependencies.initialize()
    yield
    # Shutdown: close connections cleanly
    await GlobalDependencies.close()


def create_app() -> FastAPI:
    app = FastAPI(
        title="My API",
        version="1.0.0",
        docs_url="/swagger",
        redoc_url="/api",
        lifespan=lifespan,
    )

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

    app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
    app.include_router(orders.router, prefix="/api/v1/orders", tags=["orders"])

    return app

---

APIRouter

# api/routes/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from api.dependencies import get_current_user, get_db_session
from api.schemas.users import UserResponse, CreateUserRequest
from domain.use_cases.users import CreateUserUseCase
from domain.entities.user import User

router = APIRouter()


@router.get("/", response_model=list[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 100,
    session=Depends(get_db_session),
):
    return await UserCRUD(session).list(skip=skip, limit=limit)


@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    body: CreateUserRequest,
    use_case: CreateUserUseCase = Depends(get_create_user_use_case),
):
    return await use_case.execute(body)


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: str, session=Depends(get_db_session)):
    user = await UserCRUD(session).get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: str,
    _current_user: User = Depends(get_current_user),  # requires auth
    session=Depends(get_db_session),
):
    await UserCRUD(session).delete(user_id)

---

Dependency Injection

# api/dependencies.py
from fastapi import Depends, Header, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from config.dependencies import GlobalDependencies


async def get_db_session() -> AsyncSession:
    async with GlobalDependencies.db_engine.begin() as session:
        yield session  # yields inside with-block; rolls back on exception


async def get_api_key(x_api_key: str = Header(...)) -> str:
    if x_api_key not in GlobalDependencies.valid_keys:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return x_api_key


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_db_session),
) -> User:
    payload = decode_jwt(token)               # raises 401 on bad token
    user = await UserCRUD(session).get(payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user


# Chain dependencies — get_create_user_use_case depends on get_db_session
def get_create_user_use_case(
    session: AsyncSession = Depends(get_db_session),
) -> CreateUserUseCase:
    return CreateUserUseCase(repo=UserRepo(session))

Key rules:

• yield-based dependencies run cleanup after the response is sent
• FastAPI caches dependencies within a single request — get_db_session called 3 times in one request returns the same session
• Use Depends(get_current_user) as a parameter to require auth on a route

---

Pydantic v2 Schemas

# api/schemas/users.py
from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator
from datetime import datetime
from typing import Annotated

UserId = Annotated[str, Field(min_length=1, description="User UUID")]


class CreateUserRequest(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    email: EmailStr
    role: Literal["admin", "user"] = "user"
    age: int = Field(ge=0, le=150)

    @field_validator("name")
    @classmethod
    def strip_name(cls, v: str) -> str:
        return v.strip()

    @model_validator(mode="after")
    def admin_must_have_age(self) -> "CreateUserRequest":
        if self.role == "admin" and self.age < 18:
            raise ValueError("Admins must be 18+")
        return self


class UserResponse(BaseModel):
    id: UserId
    name: str
    email: EmailStr
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)  # allows ORM → schema conversion


# Nested schemas
class OrderWithUserResponse(BaseModel):
    id: str
    total: float
    user: UserResponse             # nested
    items: list[OrderItemResponse]

---

Settings (Pydantic Settings)

# config/settings.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache


class Settings(BaseSettings):
    environment: str = "development"
    database_url: str
    redis_url: str = "redis://localhost:6379"
    secret_key: str
    allowed_origins: list[str] = ["http://localhost:3000"]

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )


@lru_cache
def get_settings() -> Settings:
    return Settings()

# In dependency:
def get_settings_dep(settings: Settings = Depends(get_settings)) -> Settings:
    return settings

---

Error Handling

# domain/exceptions.py
class ClientError(Exception):
    """400 — bad input, caller's fault"""
    def __init__(self, message: str): self.message = message

class NotFoundError(Exception):
    """404"""
    def __init__(self, resource: str, id: str):
        self.message = f"{resource} '{id}' not found"

class ServiceError(Exception):
    """500 — internal failure"""


# api/app.py — register handlers
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ClientError)
async def client_error_handler(request: Request, exc: ClientError):
    return JSONResponse(status_code=400, content={"detail": exc.message})

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
    return JSONResponse(status_code=404, content={"detail": exc.message})

@app.exception_handler(ServiceError)
async def service_error_handler(request: Request, exc: ServiceError):
    return JSONResponse(status_code=500, content={"detail": "Internal error"})

Never raise HTTPException inside use cases — only in route handlers or dependencies.

---

Middleware

# api/middleware.py
import uuid, time
from fastapi import Request

async def request_id_middleware(request: Request, call_next):
    request_id = request.headers.get("x-request-id", uuid.uuid4().hex)
    request.state.request_id = request_id
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["x-request-id"] = request_id
    response.headers["x-response-time"] = f"{duration:.3f}s"
    return response

# Register as BaseHTTPMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
app.add_middleware(BaseHTTPMiddleware, dispatch=request_id_middleware)

---

Background Tasks

from fastapi import BackgroundTasks

@router.post("/users/")
async def create_user(
    body: CreateUserRequest,
    background_tasks: BackgroundTasks,
    session=Depends(get_db_session),
):
    user = await UserCRUD(session).create(body)
    # runs after response is sent — good for emails, webhooks, cache invalidation
    background_tasks.add_task(send_welcome_email, user.email, user.name)
    return user

Use background tasks for fire-and-forget work. For durable/retryable work, use a task queue (Celery, ARQ, Temporal).

---

Authentication (JWT + OAuth2)

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt

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

@router.post("/auth/token")
async def login(form: OAuth2PasswordRequestForm = Depends(), session=Depends(get_db_session)):
    user = await authenticate_user(form.user

---

*Content truncated.*