Fastapi Patterns

FastAPI Patterns

Application Structure

app/
├── main.py              # FastAPI() instance, startup/shutdown events
├── api/
│   ├── deps.py          # Shared dependencies (db, auth, settings)
│   └── v1/
│       ├── router.py    # APIRouter for v1
│       └── users.py     # Endpoint handlers
├── models/
│   ├── user.py          # SQLAlchemy models
│   └── base.py          # Base, metadata
├── schemas/
│   ├── user.py          # Pydantic request/response schemas
│   └── common.py        # Shared schemas (pagination, errors)
├── services/
│   └── user_service.py  # Business logic (not in endpoints)
└── core/
    ├── config.py        # Settings (pydantic-settings)
    └── security.py      # Auth helpers

Dependency Injection

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

# Good: generator dependency with cleanup
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Type-annotated dependency
CurrentUser = Annotated[User, Depends(get_current_user)]

@router.get("/me", response_model=UserSchema)
async def get_me(user: CurrentUser) -> User:
    return user

Pydantic v2 Schemas

from pydantic import BaseModel, EmailStr, field_validator, model_validator, ConfigDict

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    name: str

    @field_validator("password")
    @classmethod
    def validate_password(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        return v

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # replaces orm_mode
    
    id: int
    email: EmailStr
    name: str
    created_at: datetime

Response Models and Error Handling

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

# Always specify response_model
@router.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(body: UserCreate, db: AsyncSession = Depends(get_db)) -> User:
    if await user_service.email_exists(db, body.email):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail={"code": "email_taken", "message": "Email already registered"},
        )
    return await user_service.create(db, body)

# Global exception handler
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
    return JSONResponse(status_code=400, content={"error": str(exc)})

Background Tasks

from fastapi import BackgroundTasks

@router.post("/reports")
async def generate_report(
    params: ReportParams,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
) -> dict:
    report_id = uuid4()
    # Don't fire-and-forget without error capture
    background_tasks.add_task(report_service.generate, db, report_id, params)
    return {"report_id": str(report_id), "status": "processing"}

Anti-Patterns

• Using Session (sync) inside async def endpoints
• Missing response_model on endpoints
• Business logic inside endpoint handlers (use service layer)
• async def without any actual await (just use def)
• Mutable defaults in Pydantic fields: tags: list = [] → tags: list = Field(default_factory=list)