Py Fastapi Patterns

FastAPI Patterns

Problem Statement

FastAPI API design directly affects frontend. Bad patterns here cause frontend bugs, poor developer experience, and integration issues. The OpenAPI schema drives frontend code generation.

Pattern: Dependency Injection

Problem: Repetitive code for auth, sessions, and services.

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

# ✅ CORRECT: Dependencies for common needs
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_session),
) -> User:
    user = await verify_token_and_get_user(token, session)
    if not user:
        raise HTTPException(401, "Invalid authentication")
    return user

async def get_current_active_user(
    user: User = Depends(get_current_user),
) -> User:
    if not user.is_active:
        raise HTTPException(403, "User is inactive")
    return user

# ✅ CORRECT: Endpoint using dependencies
@router.post("/assessments", response_model=AssessmentRead)
async def create_assessment(
    data: AssessmentCreate,
    current_user: User = Depends(get_current_active_user),
    session: AsyncSession = Depends(get_session),
) -> AssessmentRead:
    assessment = Assessment(**data.model_dump(), user_id=current_user.id)
    session.add(assessment)
    await session.commit()
    await session.refresh(assessment)
    return assessment

Dependency chain: get_session → get_current_user → get_current_active_user

Pattern: Response Models

Problem: Inconsistent responses, exposing internal fields, poor OpenAPI docs.

# ✅ CORRECT: Explicit response_model
@router.get("/users/{user_id}", response_model=UserRead)
async def get_user(
    user_id: UUID,
    session: AsyncSession = Depends(get_session),
) -> UserRead:
    user = await get_user_or_404(user_id, session)
    return user  # Automatically filtered to UserRead fields

# ✅ CORRECT: List response
@router.get("/users", response_model=list[UserRead])
async def list_users(...) -> list[UserRead]:
    ...

# ✅ CORRECT: Paginated response
class PaginatedResponse(SQLModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    size: int

@router.get("/assessments", response_model=PaginatedResponse[AssessmentRead])
async def list_assessments(...):
    ...

# ❌ WRONG: No response_model (exposes everything)
@router.get("/users/{user_id}")
async def get_user(user_id: UUID) -> User:  # Exposes hashed_password!
    ...

Why response_model matters:

1. Filters output to only specified fields
2. Generates accurate OpenAPI schema
3. Frontend Orval codegen depends on this

Pattern: Error Handling

Problem: Inconsistent error responses, missing context.

from fastapi import HTTPException, status

# ✅ CORRECT: Specific HTTP exceptions
@router.get("/assessments/{id}")
async def get_assessment(id: UUID, session: AsyncSession = Depends(get_session)):
    result = await session.execute(
        select(Assessment).where(Assessment.id == id)
    )
    assessment = result.scalar_one_or_none()
    
    if not assessment:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Assessment {id} not found",
        )
    
    return assessment

# ✅ CORRECT: Custom exception with handler
class AssessmentNotFoundError(Exception):
    def __init__(self, assessment_id: UUID):
        self.assessment_id = assessment_id

@app.exception_handler(AssessmentNotFoundError)
async def assessment_not_found_handler(request: Request, exc: AssessmentNotFoundError):
    return JSONResponse(
        status_code=404,
        content={
            "detail": f"Assessment {exc.assessment_id} not found",
            "error_code": "ASSESSMENT_NOT_FOUND",
        },
    )

# ✅ CORRECT: Validation error detail
@router.post("/assessments")
async def create_assessment(data: AssessmentCreate):
    if data.end_date < data.start_date:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail="end_date must be after start_date",
        )

HTTP Status Codes:

Pattern: Route Ordering

Problem: FastAPI matches first route. Order matters for overlapping paths.

# ❌ WRONG: Generic route before specific
@router.get("/users/{user_id}")  # This catches "me" as user_id!
async def get_user(user_id: str):
    ...

@router.get("/users/me")  # Never reached
async def get_current_user():
    ...

# ✅ CORRECT: Specific routes before generic
@router.get("/users/me")  # Specific first
async def get_current_user():
    ...

@router.get("/users/{user_id}")  # Generic after
async def get_user(user_id: UUID):  # UUID type also helps
    ...

Remember: Always define specific routes before generic parameterized routes.

Pattern: Path and Query Parameters

# Path parameter - required, part of URL
@router.get("/users/{user_id}")
async def get_user(user_id: UUID):  # /users/123
    ...

# Query parameters - optional, after ?
@router.get("/assessments")
async def list_assessments(
    status: str | None = None,        # /assessments?status=active
    skip: int = 0,                     # /assessments?skip=10
    limit: int = Query(default=20, le=100),  # With validation
):
    ...

# Enum for constrained values
class AssessmentStatus(str, Enum):
    DRAFT = "draft"
    ACTIVE = "active"
    COMPLETED = "completed"

@router.get("/assessments")
async def list_assessments(status: AssessmentStatus | None = None):
    ...

Pattern: Request Body Validation

from pydantic import Field, field_validator

class AssessmentCreate(SQLModel):
    title: str = Field(min_length=1, max_length=200)
    description: str | None = Field(default=None, max_length=1000)
    skill_areas: list[str] = Field(min_length=1)
    
    @field_validator("skill_areas")
    @classmethod
    def validate_skill_areas(cls, v: list[str]) -> list[str]:
        valid_areas = {"fundamentals", "advanced", "strategy"}
        for area in v:
            if area not in valid_areas:
                raise ValueError(f"Invalid skill area: {area}")
        return v

# Automatic validation - returns 422 on failure
@router.post("/assessments", response_model=AssessmentRead)
async def create_assessment(data: AssessmentCreate):
    ...

Pattern: Middleware

Problem: Cross-cutting concerns like logging, CORS, timing.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# CORS - order matters, add early
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Or ["*"] for dev
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom timing middleware
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    response.headers["X-Process-Time"] = str(duration)
    return response

# Middleware order: Last added = First executed

Pattern: OpenAPI Schema

Problem: Schema affects frontend codegen. Keep it clean.

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="API description here",
)

# Good schema descriptions
class AssessmentCreate(SQLModel):
    """Create a new skill assessment."""
    
    title: str = Field(description="Assessment title shown to user")
    skill_areas: list[str] = Field(
        description="List of skill areas to assess",
        examples=[["fundamentals", "strategy"]],
    )

# Endpoint documentation
@router.post(
    "/assessments",
    response_model=AssessmentRead,
    summary="Create assessment",
    description="Creates a new skill assessment for the current user.",
    responses={
        201: {"description": "Assessment created successfully"},
        422: {"description": "Validation error"},
    },
)
async def create_assessment(data: AssessmentCreate):
    ...

Pattern: Router Organization

# app/routers/assessments.py
from fastapi import APIRouter

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

@router.get("/")
async def list_assessments():
    ...

@router.post("/")
async def create_assessment():
    ...

# app/main.py
from app.routers import assessments, users, training

app.include_router(assessments.router, prefix="/api")
app.include_router(users.router, prefix="/api")
app.include_router(training.router, prefix="/api")

References

• OpenAPI at /docs or /openapi.json
• FastAPI documentation: https://fastapi.tiangolo.com/

Common Issues

Detection Commands

# Find endpoints missing response_model
grep -rn "@router\." --include="*.py" | grep -v "response_model"

# Find potential route ordering issues
grep -rn "@router.get" --include="*.py" | grep -E '"/\w+/\{|"/\w+/\w+"'

# Check OpenAPI schema
curl http://localhost:8000/openapi.json | jq '.paths'