FastAPI Patterns & Best Practices
Project Structure
app/
├── main.py # FastAPI app initialization
├── config.py # Settings and configuration
├── dependencies.py # Shared dependencies
├── routers/
│ ├── __init__.py
│ ├── users.py
│ └── items.py
├── models/
│ ├── __init__.py
│ ├── user.py # SQLAlchemy models
│ └── item.py
├── schemas/
│ ├── __init__.py
│ ├── user.py # Pydantic schemas
│ └── item.py
├── services/
│ ├── __init__.py
│ ├── user_service.py
│ └── item_service.py
├── repositories/
│ ├── __init__.py
│ └── user_repository.py
└── utils/
├── __init__.py
└── security.py
Configuration
Settings with Pydantic
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
app_name: str = "My API"
debug: bool = False
database_url: str
redis_url: str
secret_key: str
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
Dependency Injection
Database Session
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from typing import AsyncGenerator
engine = create_async_engine(settings.database_url, echo=settings.debug)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
Current User Dependency
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = await db.get(User, user_id)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user)
) -> User:
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
Routers
User Router
from fastapi import APIRouter, Depends, HTTPException, status
from typing import Annotated
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/", response_model=list[UserOut])
async def list_users(
db: Annotated[AsyncSession, Depends(get_db)],
skip: int = 0,
limit: int = Query(default=100, le=100),
current_user: Annotated[User, Depends(get_current_active_user)] = None,
):
users = await user_service.get_users(db, skip=skip, limit=limit)
return users
@router.post("/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(
user_in: UserCreate,
db: Annotated[AsyncSession, Depends(get_db)],
):
db_user = await user_service.get_by_email(db, email=user_in.email)
if db_user:
raise HTTPException(
status_code=400,
detail="Email already registered"
)
return await user_service.create(db, user_in)
Schemas
Request/Response Schemas
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
class UserBase(BaseModel):
email: EmailStr
full_name: str | None = None
class UserCreate(UserBase):
password: str = Field(min_length=8)
class UserUpdate(BaseModel):
email: EmailStr | None = None
full_name: str | None = None
password: str | None = Field(default=None, min_length=8)
class UserOut(UserBase):
id: int
is_active: bool
created_at: datetime
model_config = {"from_attributes": True}
Error Handling
Custom Exception Handlers
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class AppException(Exception):
def __init__(self, status_code: int, detail: str, code: str):
self.status_code = status_code
self.detail = detail
self.code = code
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
return JSONResponse(
status_code=exc.status_code,
content={
"error": {
"code": exc.code,
"message": exc.detail,
"path": str(request.url)
}
}
)
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={
"error": {
"code": "INTERNAL_ERROR",
"message": "An unexpected error occurred"
}
}
)
Background Tasks
Async Background Tasks
from fastapi import BackgroundTasks
async def send_email(email: str, subject: str, body: str):
# Async email sending logic
await email_client.send(to=email, subject=subject, body=body)
@router.post("/users/")
async def create_user(
user: UserCreate,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db)
):
db_user = await user_service.create(db, user)
background_tasks.add_task(
send_email,
email=user.email,
subject="Welcome!",
body="Thanks for signing up"
)
return db_user
Middleware
Request ID Middleware
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
CORS Middleware
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://frontend.example.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Testing
Test Client
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
@pytest.fixture
async def async_client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def test_db():
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSession(engine) as session:
yield session
async def test_create_user(async_client, test_db):
response = await async_client.post(
"/users/",
json={"email": "test@example.com", "password": "securepassword123"}
)
assert response.status_code == 201
data = response.json()
assert data["email"] == "test@example.com"
Production
Uvicorn with Gunicorn
gunicorn app.main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 120 \
--keep-alive 5
Health Check Endpoint
from fastapi import APIRouter
router = APIRouter(tags=["health"])
@router.get("/health")
async def health_check(db: AsyncSession = Depends(get_db)):
try:
await db.execute("SELECT 1")
return {"status": "healthy", "database": "connected"}
except Exception:
return JSONResponse(
status_code=503,
content={"status": "unhealthy", "database": "disconnected"}
)
advisely