FastAPI Patterns
Modern FastAPI 0.110+ patterns and best practices with Pydantic v2.
Project Structure
app/ ├── main.py # App factory, middleware, startup ├── config.py # Settings via pydantic-settings ├── dependencies.py # Shared dependencies ├── models/ # SQLAlchemy / SQLModel models │ └── user.py ├── schemas/ # Pydantic request/response schemas │ └── user.py ├── routers/ # Route handlers (one per domain) │ └── users.py ├── services/ # Business logic layer │ └── user_service.py ├── repositories/ # Data access layer │ └── user_repo.py └── tests/ ├── conftest.py └── test_users.py
Dependency Injection
Database Session
from collections.abc import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
engine = create_async_engine(settings.DATABASE_URL) async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: try: yield session await session.commit() except Exception: await session.rollback() raise
Service Dependencies
from fastapi import Depends
async def get_user_service( db: AsyncSession = Depends(get_db), cache: Redis = Depends(get_redis), ) -> UserService: return UserService(db=db, cache=cache)
@router.get("/users/{user_id}") async def get_user( user_id: UUID, service: UserService = Depends(get_user_service), ) -> UserResponse: return await service.get_by_id(user_id)
Authentication Dependency
from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
async def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), db: AsyncSession = Depends(get_db), ) -> User: token = credentials.credentials payload = verify_jwt(token) # Raises on invalid user = await db.get(User, payload["sub"]) if not user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) return user
Pydantic v2 Schemas
Request/Response Separation
from pydantic import BaseModel, ConfigDict, EmailStr, Field from uuid import UUID from datetime import datetime
class UserCreate(BaseModel): name: str = Field(min_length=1, max_length=100) email: EmailStr
class UserUpdate(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=100) email: EmailStr | None = None
class UserResponse(BaseModel): model_config = ConfigDict(from_attributes=True)
id: UUID
name: str
email: str
created_at: datetime
Validated Query Parameters
from pydantic import BaseModel, Field
class PaginationParams(BaseModel): page: int = Field(default=1, ge=1) per_page: int = Field(default=20, ge=1, le=100)
@property
def offset(self) -> int:
return (self.page - 1) * self.per_page
@router.get("/users") async def list_users( params: PaginationParams = Depends(), service: UserService = Depends(get_user_service), ) -> list[UserResponse]: return await service.list(offset=params.offset, limit=params.per_page)
Async Patterns
Concurrent External Calls
import asyncio
@router.get("/dashboard") async def dashboard(user: User = Depends(get_current_user)): stats, notifications, recent = await asyncio.gather( fetch_user_stats(user.id), fetch_notifications(user.id), fetch_recent_activity(user.id), ) return {"stats": stats, "notifications": notifications, "recent": recent}
Background Tasks
from fastapi import BackgroundTasks
async def send_welcome_email(email: str, name: str) -> None: # Long-running task runs after response is sent await email_service.send(to=email, template="welcome", context={"name": name})
@router.post("/users", status_code=201) async def create_user( body: UserCreate, background: BackgroundTasks, service: UserService = Depends(get_user_service), ) -> UserResponse: user = await service.create(body) background.add_task(send_welcome_email, user.email, user.name) return user
Middleware
import time from fastapi import FastAPI, Request from starlette.middleware.cors import CORSMiddleware
app = FastAPI()
CORS
app.add_middleware( CORSMiddleware, allow_origins=settings.ALLOWED_ORIGINS, allow_methods=[""], allow_headers=[""], )
Request timing
@app.middleware("http") async def add_timing_header(request: Request, call_next): start = time.perf_counter() response = await call_next(request) elapsed = time.perf_counter() - start response.headers["X-Process-Time"] = f"{elapsed:.4f}" return response
Error Handling
from fastapi import HTTPException, Request from fastapi.responses import JSONResponse
class AppError(Exception): def init(self, message: str, status_code: int = 400): self.message = message self.status_code = status_code
@app.exception_handler(AppError) async def app_error_handler(request: Request, exc: AppError) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content={"detail": exc.message}, )
Configuration with pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", case_sensitive=False, )
DATABASE_URL: str
REDIS_URL: str = "redis://localhost:6379"
SECRET_KEY: str
DEBUG: bool = False
ALLOWED_ORIGINS: list[str] = ["http://localhost:3000"]
settings = Settings()
Testing
import pytest from httpx import AsyncClient, ASGITransport from app.main import app
@pytest.fixture async def client(): transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as ac: yield ac
@pytest.mark.anyio async def test_create_user(client: AsyncClient): response = await client.post("/users", json={"name": "Alice", "email": "a@b.com"}) assert response.status_code == 201 data = response.json() assert data["name"] == "Alice"
Checklist
-
Async endpoints for all I/O operations
-
Dependency injection for DB sessions, services, auth
-
Pydantic v2 schemas with from_attributes=True
-
Separate request/response models (never expose DB models)
-
Background tasks for email, webhooks, logging
-
CORS middleware configured
-
Exception handlers for custom error types
-
Settings via pydantic-settings (not raw os.environ)
-
Tests use httpx AsyncClient with ASGITransport