Peopleforrester Claude Dotfiles Fastapi Patterns

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

Converted and distributed by TomeVault — claim your Tome and manage your conversions.