FastAPI Development Patterns

Router Organization

Split routes by domain using APIRouter. Mount all routers in a single app.py / main.py.

# api/users.py
from fastapi import APIRouter, Depends
from .dependencies import get_db

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}")
async def get_user(user_id: int, db=Depends(get_db)):
    ...

# main.py
from fastapi import FastAPI
from .api import users, orders, reports

app = FastAPI()
app.include_router(users.router)
app.include_router(orders.router)
app.include_router(reports.router)

File naming: one router variable per file, file name matches the resource (users.py, orders.py).

Request & Response Models

Use separate Pydantic BaseModel classes for request and response. Do not reuse the same model for both.

from pydantic import BaseModel, Field
from typing import Optional

class CreateUserRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., pattern=r"^[^@]+@[^@]+\.[^@]+$")
    role: str = Field(default="viewer")

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    role: str

    model_config = {"from_attributes": True}  # enables ORM mode

Naming convention: XxxRequest for input, XxxResponse for output, XxxSchema for shared internal use.

Dependency Injection

Use Depends() for any shared resource: database sessions, cache clients, auth context, or external service clients. The same pattern works for any injectable singleton.

# dependencies.py
from functools import lru_cache
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
import redis.asyncio as aioredis
from .config import Settings

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

engine = create_async_engine(get_settings().database_url)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

async def get_redis() -> aioredis.Redis:
    return aioredis.from_url(get_settings().redis_url)

# The same pattern applies to any singleton client — LLM client, HTTP client, etc.:
# async def get_llm_client() -> LLMClient:
#     return LLMClient(api_key=get_settings().llm_api_key)

# Usage in route
@router.post("/orders", response_model=OrderResponse)
async def create_order(
    request: CreateOrderRequest,
    db: AsyncSession = Depends(get_db),
    cache: aioredis.Redis = Depends(get_redis),
):
    ...

Exception Handling

Define custom exceptions and register global handlers. Never return raw exception details to the client.

# exceptions.py
class AppError(Exception):
    def __init__(self, code: str, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code

class NotFoundError(AppError):
    def __init__(self, resource: str, resource_id):
        super().__init__("NOT_FOUND", f"{resource} {resource_id} not found", 404)

# main.py — register handler
from fastapi import Request
from fastapi.responses import JSONResponse

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

# In route — raise, never return error dict manually
@router.get("/{user_id}")
async def get_user(user_id: int, db=Depends(get_db)):
    user = await user_service.get(db, user_id)
    if not user:
        raise NotFoundError("User", user_id)
    return UserResponse.model_validate(user)

HTTP Status Codes

Declare the response model and status code explicitly on the route decorator:

@router.post("/users", response_model=UserResponse, status_code=201)
async def create_user(request: CreateUserRequest, db=Depends(get_db)):
    ...

Async Patterns

Use async def for all route functions. Use await for any I/O operation (DB, cache, HTTP calls).

# GOOD — async all the way down
@router.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

# BAD — blocking call inside async route blocks the event loop
@router.get("/users/{user_id}")
async def get_user(user_id: int):
    time.sleep(1)          # blocks event loop
    result = requests.get(...)  # sync HTTP inside async context

When wrapping a synchronous CPU-bound function, use asyncio.to_thread() to avoid blocking:

import asyncio

@router.post("/analyze")
async def analyze(payload: AnalyzeRequest):
    result = await asyncio.to_thread(heavy_cpu_function, payload.data)
    return {"result": result}