FastAPI production patterns — Pydantic models, dependency injection, async DB, middleware, auth (OAuth2/JWT), background tasks, error handling, OpenAPI customization, and pytest-asyncio testing. Use when building or reviewing FastAPI services.

FastAPI Patterns

Production patterns for FastAPI — asynchronous APIs with strong typing via Pydantic.

When to activate

Creating new FastAPI endpoints
Adding DI, middleware, authentication
Structuring a new FastAPI service from scratch
Reviewing async/await usage in endpoints
Integrating with async SQLAlchemy / SQLModel / Tortoise

Recommended project structure

app/
├── main.py              # FastAPI instance + routers include
├── core/
│   ├── config.py        # Settings via BaseSettings
│   ├── security.py      # OAuth2PasswordBearer, JWT, password hash
│   └── deps.py          # Shared dependencies (get_current_user, get_db)
├── api/
│   └── v1/
│       ├── routes/
│       │   ├── users.py
│       │   └── items.py
│       └── router.py    # APIRouter aggregation
├── models/              # SQLAlchemy models
├── schemas/             # Pydantic request/response schemas
├── crud/                # DB access functions (pure)
└── db/
    ├── session.py       # async engine + sessionmaker
    └── base.py          # declarative base

Pydantic Models (v2)

Separate request, response and DB models:

from pydantic import BaseModel, EmailStr, Field
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    full_name: str | None = Field(None, max_length=120)

class UserCreate(UserBase):
    password: str = Field(min_length=12)

class UserUpdate(BaseModel):
    full_name: str | None = None
    password: str | None = None

class UserOut(UserBase):
    id: int
    created_at: datetime
    model_config = {"from_attributes": True}  # v2: allows ORM → schema

Rules:

UserCreate has a password, UserOut never does
Use EmailStr, HttpUrl, AnyUrl instead of str
model_config = {"from_attributes": True} replaces orm_mode = True (v2)
Field(...) for constraints (length, regex, ge/le)

Dependency Injection

DI is FastAPI's central mechanism. Use it for:

# app/core/deps.py
from typing import Annotated
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    async with SessionLocal() as session:
        yield session

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    user = await decode_token_and_load(token, db)
    if not user:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED)
    return user

# Usage in a route
@router.get("/me")
async def read_me(
    current_user: Annotated[User, Depends(get_current_user)],
) -> UserOut:
    return current_user

Pattern: use Annotated[Type, Depends(...)] (PEP 593) instead of = Depends(...) — clearer and enables reuse.

Async DB (SQLAlchemy 2.0)

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine(settings.DATABASE_URL, pool_size=20, max_overflow=10)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

# Query
async def get_user(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

Pitfalls:

expire_on_commit=False is critical in async — otherwise accessing attributes after commit triggers a synchronous lazy-load that blocks the loop
Never mix Session (sync) and AsyncSession in the same request
scalar_one_or_none() > first() for clarity

Error Handling

from fastapi import HTTPException, status
from fastapi.responses import JSONResponse
from fastapi.requests import Request

class BusinessError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message}},
    )

Pattern: use custom exceptions for domain errors; HTTPException only for generic HTTP errors.

Authentication (OAuth2 + JWT)

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from jose import jwt, JWTError

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(subject: str, expires_delta: timedelta) -> str:
    payload = {"sub": subject, "exp": datetime.utcnow() + expires_delta}
    return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")