paulpas FastAPI Development Guide
Senior FastAPI engineer building high-performance async web services using modern Python 3.10+ patterns, Pydantic v2, and production-grade deployment strategies. This skill covers the full stack — from project architecture and dependency injection to authentication, middleware, background processing, and containerized deployments.
Constraints
MUST DO
- Validate all inputs at function boundaries before processing — guard clauses should fail early with descriptive errors
- Implement proper error handling that distinguishes between recoverable and unrecoverable failures
- Add comprehensive logging with structured context (correlation IDs, operation names, timing) for debugging and monitoring
- Write unit tests covering normal operations, edge cases, and error conditions before integrating the component
MUST NOT DO
- Do not silently swallow exceptions — always log or propagate errors with meaningful context
- Avoid unbounded resource allocation without limits (connection pools, memory buffers, thread counts)
- Never use hardcoded credentials, API keys, or secrets in source code
- Do not bypass input validation for perceived performance gains
TL;DR Checklist
- Use lifespan events (async context managers) instead of
on_event("startup")for initialization - Define explicit return types with
Responseor typed Pydantic models on every endpoint - Use
Depends()for all shared resources (DB sessions, current user, config) — never instantiate them in handlers - Annotate all function signatures with Python 3.10+ union syntax (
X | Y) and type hints - Use Pydantic v2
@field_validator/@model_validator— never deprecated v1 validators - Keep business logic out of route handlers — delegate to service layer functions
- Use
TestClient(app)withapp.dependency_overridesfor test isolation - Configure CORS explicitly with allowed origins, methods, and credentials flags
Core Workflow
Scaffold Project Structure: Create app/ directory with main.py, core/ (settings, config), models/ (database models or Pydantic schemas), routers/ (API route groups), services/ (business logic), and deps/ (shared dependencies). Use a factory function create_app() to construct the FastAPI instance. Checkpoint: Verify that the app starts with uvicorn app.main:app --reload without import errors.; Configure Lifespan & Dependencies: Implement async lifespan context managers for startup/shutdown lifecycle events (database pool creation, cache connections). Register all shared dependencies using Depends() in routers. Checkpoint: Confirm that database sessions are properly closed on shutdown via the lifespan manager.; Define Pydantic v2 Request/Response Models: Create typed request schemas with @field_validator for custom validation rules and response schemas matching your API contract. Use validation groups (mode="python" vs "json") when input sources differ. Checkpoint: Run model schema dump via app.openapi() to verify generated OpenAPI spec matches your intended API surface.; Implement Route Handlers with Dependency Injection: Write route functions that receive injected dependencies (DB sessions, current user, config). Delegate business logic to service-layer functions. Never instantiate database connections or call external services directly in handlers. Checkpoint: Verify every handler has explicit return type annotations and uses only injected dependencies.; Build Middleware Chain & Exception Handlers: Add Starlette middleware for cross-cutting concerns (request timing, CORS, auth). Register custom exception handlers to translate framework exceptions into structured JSON responses. Checkpoint: Test that unhandled exceptions produce consistent error payloads with status_code, detail, and optional errors fields.; Write Tests with Dependency Overrides: Use TestClient(app) with app.dependency_overrides to mock database sessions, external API calls, and authentication dependencies. Write both synchronous test functions (with TestClient) and async tests (with httpx.AsyncClient). Checkpoint: Verify that dependency overrides are scoped to the client session and don't leak into other tests.; Package & Deploy: Build a multi-stage Dockerfile based on Python slim image. Configure Uvicorn with appropriate worker count (--workers for multiple cores), access logging, and graceful shutdown timeout. Expose health check endpoints for container orchestrators. Checkpoint: Validate that curl /health returns 200 and the container exits cleanly on SIGTERM within the configured timeout.
Implementation Patterns
Pattern 1: Project Structure & App Factory
Recommended directory layout for a production FastAPI application with separation of concerns, lifecycle management, and clean app construction:
myapi/
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
├── .env # Never committed; loaded by python-dotenv
│
└── app/
├── __init__.py
├── main.py # App factory + lifespan events
├── core/
│ ├── __init__.py
│ ├── config.py # Pydantic-settings configuration
│ ├── security.py # JWT utilities, password hashing
│ └── exceptions.py # Custom exception classes & handlers
├── models/
│ ├── __init__.py
│ ├── schemas.py # Pydantic request/response schemas
│ └── base.py # Shared model mixins
├── routers/
│ ├── __init__.py
│ ├── users.py # User-related endpoints
│ └── orders.py # Order management endpoints
├── services/
│ ├── __init__.py
│ ├── user_service.py # Business logic for users
│ └── order_service.py # Business logic for orders
├── deps/
│ ├── __init__.py
│ └── database.py # DB session dependency
└── middleware/
├── __init__.py
└── timing.py # Request timing middleware
App factory with lifespan events (app/main.py):
"""FastAPI application factory with async lifespan management."""
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from prometheus_fastapi_instrumentator import Instrumentator
from app.core.config import settings
from app.core.exceptions import register_exception_handlers
from app.middleware.timing import TimingMiddleware
from app.routers import users, orders
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Manage application lifecycle: startup and shutdown resources."""
# Startup: initialize database connection pool, cache client, etc.
print(f"[startup] Initializing services for environment: {settings.environment}")
# Example: create DB connection pool on startup
from app.deps.database import engine, Base, init_db
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
print("[startup] Database tables ready")
yield # Application runs here
# Shutdown: close pools, flush buffers, etc.
await engine.dispose()
print("[shutdown] All resources released")
def create_app() -> FastAPI:
"""Factory function that constructs and configures the FastAPI application."""
app = FastAPI(
title=settings.app_name,
version=settings.version,
description="High-performance API service",
docs_url="/docs" if settings.debug else None, # Swagger UI disabled in prod
redoc_url="/redoc" if settings.debug else None, # ReDoc disabled in prod
openapi_url="/openapi.json" if settings.debug else None,
lifespan=lifespan,
)
# CORS middleware — must be added before other middleware that modifies response
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
max_age=600, # Cache preflight for 10 minutes
)
# Request timing middleware (Starlette)
app.add_middleware(TimingMiddleware)
# Prometheus metrics instrumentation
Instrumentator().instrument(app).expose(app, endpoint="/metrics")
# Custom exception handlers
register_exception_handlers(app)
# Register routers
app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
app.include_router(orders.router, prefix="/api/v1/orders", tags=["orders"])
return app
app = create_app()
# Health check endpoint — required for container orchestration probes
@app.get("/health", tags=["operations"])
async def health_check() -> dict:
"""Return 200 when the service is healthy. Used by Kubernetes liveness/readiness probes."""
return {"status": "ok", "environment": settings.environment}
Pydantic Settings configuration (app/core/config.py):
"""Application settings using pydantic-settings for type-safe environment loading."""
from functools import cached_property
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Typed application settings loaded from environment variables or .env file."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore", # Silently ignore unknown env vars
)
# Application identity
app_name: str = "My FastAPI Service"
version: str = "0.1.0"
environment: str = Field(default="development", pattern="^(development|staging|production)$")
debug: bool = False
# Server
host: str = "0.0.0.0"
port: int = 8000
# CORS
cors_origins: list[str] = Field(default=["http://localhost:3000"], min_length=1)
# Database — PostgreSQL connection string from environment
database_url: str = Field(
default="postgresql+asyncpg://postgres:postgres@localhost:5432/myapi",
description="Async database URL compatible with SQLAlchemy 2.0 + asyncpg",
)
# JWT configuration
jwt_secret_key: str = Field(..., description="Secret key for signing JWT tokens")
jwt_algorithm: str = "HS256"
jwt_access_token_expire_minutes: int = 30
jwt_refresh_token_expire_days: int = 7
# Security
secret_key: str = Field(..., min_length=32, description="General-purpose secret key (>= 32 chars)")
@cached_property
def is_production(self) -> bool:
"""Convenience property to check if running in production environment."""
return self.environment == "production"
# Singleton settings instance — loaded once at import time
settings = Settings()
Pattern 2: Dependency Injection System
FastAPI's dependency injection is its killer feature. It enables testable, reusable, and cleanly-scoped resource management. This pattern covers function deps, class-based deps with setup/teardown, scanning (nested depends), override for testing, and caching behavior.
# app/deps/database.py — Database session dependency with proper cleanup
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.core.config import settings
engine = create_async_engine(
settings.database_url,
echo=settings.debug,
pool_size=10,
max_overflow=20,
)
async_session_factory = async_sessionmaker(
bind=engine,
class_=AsyncSession,
expire_on_commit=False, # Prevents deferred load issues after commit
)
async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
"""Yield an async database session and guarantee cleanup on exit.
This is the canonical dependency for injecting DB sessions into endpoints.
The session is rolled back on exception and closed regardless of outcome.
"""
session: AsyncSession = async_session_factory()
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
Class-based dependency with setup/teardown (app/deps/cache.py):
# app/deps/cache.py — Cache connection with lifecycle management
from collections.abc import AsyncGenerator
import redis.asyncio as aioredis
from fastapi import Depends
from app.core.config import settings
class CacheDependency:
"""Class-based dependency for managing Redis connection lifecycle.
__init__ runs once at construction (fast, no I/O).
__call__ runs per-request and returns the actual resource.
Async cleanup in the generator handles teardown.
"""
def __init__(self) -> None:
# Only store config — do NOT open connections here
self._url = f"redis://:{settings.redis_password}@{settings.redis_host}:{settings.redis_port}/0" # noqa: E501
async def __call__(self) -> AsyncGenerator[aioredis.Redis, None]:
"""Create Redis connection per-request and close on exit."""
client: aioredis.Redis = aioredis.from_url(
self._url,
encoding="utf-8",
decode_responses=True,
)
try:
yield client
finally:
await client.aclose()
# Singleton instance — reused across requests for efficiency
cache_dep = CacheDependency()
Dependency scanning with nested Depends and caching (app/routers/users.py):
# app/routers/users.py — Endpoints using dependency scanning
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import EmailStr
from app.deps.database import get_db_session
from app.deps.cache import cache_dep
from app.models.schemas import UserCreate, UserRead, UserUpdate
from app.services.user_service import (
create_user as _create_user,
get_user_by_id as _get_user_by_id,
update_user as _update_user,
)
from app.core.security import get_current_active_user
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/{user_id}", response_model=UserRead, summary="Get user by ID")
async def read_user(
user_id: int,
db=Depends(get_db_session),
cache=Depends(cache_dep),
) -> UserRead:
"""Fetch a user, checking the cache first via dependency scanning."""
# Dependency scanning: get both db session and cache instance
cache_key = f"user:{user_id}"
cached = await cache.aget(cache_key)
if cached is not None:
from pydantic.json import model_validate_json
return model_validate_json(UserRead, cached) # Fast path: return cached
user = await _get_user_by_id(db, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# Cache the result for 5 minutes
import json
await cache.setex(cache_key, 300, UserRead.model_dump_json(user))
return user
@router.post("/", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user_endpoint(
user_data: UserCreate,
db=Depends(get_db_session),
) -> UserRead:
"""Create a new user with validated input from Pydantic model."""
# Business logic is in the service layer — never inline in handlers
user = await _create_user(db, user_data)
# Invalidate any existing cache for this email
cache = await cache_dep.__call__() # Get fresh instance for invalidation
async with cache:
await cache.delete(f"user:{user.email}")
return user
@router.put("/{user_id}", response_model=UserRead)
async def update_user_endpoint(
user_id: int,
updates: UserUpdate,
current_user=Depends(get_current_active_user), # Auth dependency
db=Depends(get_db_session),
) -> UserRead:
"""Update an existing user. Only the owner or admin can modify."""
updated = await _update_user(db, user_id, updates)
if updated is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
return updated
Dependency override for testing (tests/conftest.py):
# tests/conftest.py — Test fixtures with dependency injection overrides
import pytest
from collections.abc import AsyncGenerator
from fastapi.testclient import TestClient
from app.main import create_app
@pytest.fixture(scope="session")
def app() -> FastAPI:
"""Session-scoped test application instance."""
return create_app()
@pytest.fixture
def client(app: FastAPI) -> AsyncGenerator[TestClient, None]:
"""Test client with dependency overrides for isolated testing.
Overrides replace real database sessions and cache connections with
in-memory test doubles. Dependencies are scoped to this client only.
"""
# Override the database session dependency
async def override_get_db():
"""Provide an in-memory SQLite session for testing."""
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from app.models.base import Base
test_engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
session_factory = async_sessionmaker(bind=test_engine, class_=AsyncSession, expire_on_commit=False) # noqa: E501
session: AsyncSession = session_factory()
try:
yield session
finally:
await session.close()
await test_engine.dispose()
from app.main import app as real_app
real_app.dependency_overrides[get_db_session] = override_get_db
real_app.dependency_overrides[cache_dep.__call__] = lambda: _mock_cache() # type: ignore[arg-type]
with TestClient(real_app) as test_client:
yield test_client
# Clean overrides after each test
real_app.dependency_overrides.clear()
BAD vs. GOOD: When NOT to over-engineer DI:
# ❌ BAD: Over-engineering — creating a dependency for something that is just a constant
class SettingsDependency:
"""Don't do this. Settings are already a module-level singleton."""
def __call__(self) -> Settings:
return settings # Unnecessary class wrapper
def bad_endpoint(settings: Settings = Depends(SettingsDependency())) -> dict:
return {"env": settings.environment}
# ✅ GOOD: Use the singleton directly — no dependency wrapper needed
def good_endpoint() -> dict:
from app.core.config import settings
return {"env": settings.environment}
# ❌ BAD: Inline business logic that should be in a service layer
async def bad_handler(db: AsyncSession = Depends(get_db_session)) -> dict:
# This is 20 lines of business logic crammed into a route handler — unmaintainable
users = await db.execute(select(User).where(User.is_active == True))
active_users = [u for u in users.scalars().all() if u.last_login > days_ago(30)]
result = {"count": len(active_users), "recent": [...]}
return result
# ✅ GOOD: Delegate to a service function — handler is thin and declarative
async def good_handler(db: AsyncSession = Depends(get_db_session)) -> dict:
active_count, recent_list = await user_service.get_active_users_last_30_days(db)
return {"count": active_count, "recent": recent_list}
Pattern 3: Pydantic v2 Model Patterns
Pydantic v2 is a fundamental rewrite from v1 with breaking changes. Key differences: model_config replaces class Config, @field_validator replaces @validator, and validation groups (mode="python" vs "json") handle different input contexts.
# app/models/schemas.py — Pydantic v2 request and response schemas
from datetime import date, datetime
from enum import Enum
from typing import Any
from pydantic import (
BaseModel,
ConfigDict,
EmailStr,
Field,
field_validator,
model_validator,
)
class UserRole(str, Enum):
"""Valid user roles in the system."""
ADMIN = "admin"
USER = "user"
VIEWER = "viewer"
class UserCreate(BaseModel):
"""Schema for creating a new user. Used for POST /users request body."""
model_config = ConfigDict(
json_schema_extra={
"examples": [
{
"email": "alice@example.com",
"username": "alice",
"password": "SecureP@ss123!",
"full_name": "Alice Developer",
},
],
},
)
email: EmailStr
username: str = Field(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
password: str = Field(min_length=8, max_length=128)
full_name: str = Field(min_length=1, max_length=200)
@field_validator("username")
@classmethod
def username_no_leading_trailing(cls, v: str) -> str:
stripped = v.strip()
if stripped != v:
raise ValueError("username must not have leading or trailing whitespace")
return stripped
@field_validator("password")
@classmethod
def password_complexity(cls, v: str) -> str:
"""Enforce password complexity: at least one uppercase, one lowercase, one digit."""
if not any(c.isupper() for c in v):
raise ValueError("password must contain at least one uppercase letter")
if not any(c.islower() for c in v):
raise ValueError("password must contain at least one lowercase letter")
if not any(c.isdigit() for c in v):
raise ValueError("password must contain at least one digit")
return v
class UserUpdate(BaseModel):
"""Schema for updating a user. All fields optional for PATCH-style updates."""
email: EmailStr | None = None
full_name: str | None = Field(default=None, min_length=1, max_length=200)
role: UserRole | None = None
@model_validator(mode="before")
@classmethod
def reject_blank_update(cls, data: Any) -> Any:
"""Reject requests where all fields are None or empty — nothing to update."""
if isinstance(data, dict):
if not any(v is not None and v != "" for v in data.values()):
raise ValueError("At least one field must be provided for update")
return data
class UserRead(BaseModel):
"""Schema for reading user data. Used for GET /users/{id} response."""
model_config = ConfigDict(from_attributes=True) # SQLAlchemy ORM-compatible
id: int
email: EmailStr
username: str
full_name: str
role: UserRole
is_active: bool
created_at: datetime
updated_at: datetime | None = None
class PasswordChange(BaseModel):
"""Schema for changing password with current and new password validation."""
current_password: str
new_password: str = Field(min_length=8, max_length=128)
@model_validator(mode="after")
def passwords_must_differ(self) -> "PasswordChange":
"""Ensure the new password is different from the current one."""
if self.current_password == self.new_password:
raise ValueError("new_password must differ from current_password")
return self
Nested models with validation groups:
# app/models/schemas.py — Nested request/response models
from pydantic import ValidationError
class Address(BaseModel):
"""Address nested model used in user profile and order contexts."""
street: str = Field(min_length=5, max_length=300)
city: str = Field(min_length=2, max_length=100)
state: str = Field(min_length=2, max_length=50)
postal_code: str = Field(pattern=r"^\d{5}(-\d{4})?$") # US ZIP+4 format
country: str = Field(default="US", pattern=r"^[A-Z]{2}$")
class OrderCreate(BaseModel):
"""Schema for creating an order with nested address."""
items: list[dict[str, Any]] = Field(min_length=1, description="List of items to order")
shipping_address: Address
billing_address: Address | None = None
notes: str | None = Field(default=None, max_length=500)
class OrderResponse(BaseModel):
"""Full order response including nested address data."""
id: int
status: str
total_amount: float
items_count: int
shipping_address: Address
billing_address: Address | None = None
created_at: datetime
model_config = ConfigDict(from_attributes=True)
# Validation groups example — validate differently based on input source
class ProductCreate(BaseModel):
"""Product creation schema with validation groups."""
name: str
price: float
description: str | None = None
@field_validator("price", mode="before")
@classmethod
def ensure_positive(cls, v: float) -> float:
"""Validate that price is positive — used for both JSON and Python input."""
if v < 0:
raise ValueError("price must be non-negative")
return v
@field_validator("price", mode="after")
@classmethod
def round_price(cls, v: float) -> float:
"""Round price to 2 decimal places for currency precision."""
return round(v, 2)
# JSON Schema output example — FastAPI uses this for OpenAPI documentation
def dump_json_schema() -> dict:
"""Dump the generated JSON schema for external consumption (e.g., API clients)."""
return OrderCreate.model_json_schema(mode="json")
# Example output:
# {
# "$defs": {
# "Address": {
# "properties": {
# "street": {"maxLength": 300, "minLength": 5, "title": "Street", "type": "string"},
# "city": {"maxLength": 100, "minLength": 2, "title": "City", "type": "string"},
# ...
# },
# "required": ["street", "city", "state", "postal_code"],
# "title": "Address",
# "type": "object"
# }
# },
# "properties": {
# "items": {"items": {"type": "object"}, "minItems": 1, "title": "Items", "type": "array"},
# "shipping_address": {"$ref": "#/$defs/Address"},
# ...
# },
# "required": ["items", "shipping_address"],
# "title": "OrderCreate",
# "type": "object"
# }
Pattern 4: Authentication & Authorization
JWT-based authentication with OAuth2 password flow, role-based access control, and token refresh. This pattern shows complete authentication infrastructure.
# app/core/security.py — JWT utilities and OAuth2 password flow
import base64
import hashlib
from datetime import datetime, timedelta, timezone
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from app.core.config import settings
# Password hashing — bcrypt with auto-rounds
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# OAuth2 token URL dependency — FastAPI uses this to generate the Swagger UI flow
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str:
"""Create a JWT access token with configurable expiration.
Args:
subject: The identifier to encode in the token (e.g., user ID or email).
expires_delta: Optional custom expiry. Defaults to settings value.
Returns:
Signed JWT token string.
"""
if expires_delta is None:
expires_delta = timedelta(minutes=settings.jwt_access_token_expire_minutes)
expire = datetime.now(timezone.utc) + expires_delta
payload = {
"sub": subject, # Subject — typically user identifier
"exp": expire,
"iat": datetime.now(timezone.utc),
"type": "access",
}
return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
def create_refresh_token(subject: str) -> str:
"""Create a long-lived refresh token for obtaining new access tokens."""
expire = datetime.now(timezone.utc) + timedelta(days=settings.jwt_refresh_token_expire_days)
payload = {
"sub": subject,
"exp": expire,
"iat": datetime.now(timezone.utc),
"type": "refresh",
}
return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
def decode_token(token: str) -> dict:
"""Decode and verify a JWT token. Raises HTTPException on failure."""
try:
payload = jwt.decode(
token,
settings.jwt_secret_key,
algorithms=[settings.jwt_algorithm],
)
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
)
except jwt.InvalidTokenError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
)
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify a plain password against its bcrypt hash."""
return pwd_context.verify(plain_password, hashed_password)
def hash_password(password: str) -> str:
"""Hash a plaintext password using bcrypt with auto-calculated rounds."""
return pwd_context.hash(password)
Auth dependency for extracting the current user:
# app/core/auth.py — Authentication & authorization dependencies
from fastapi import Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.security import decode_token, oauth2_scheme
from app.deps.database import get_db_session
from app.models.base import User # SQLAlchemy user model
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db_session),
) -> User:
"""Extract and validate the current user from the Authorization Bearer token.
Returns None if the token is invalid (endpoint decides whether to reject).
Use get_current_active_user() for endpoints that require authentication.
"""
payload = decode_token(token)
subject: str = payload.get("sub")
if subject is None or payload.get("type") != "access":
return None
result = await db.execute(select(User).where(User.id == int(subject)))
user = result.scalar_one_or_none()
return user
async def get_current_active_user(
current_user: User | None = Depends(get_current_user),
) -> User:
"""Require an authenticated, active user. Rejects unauthenticated requests."""
if current_user is None or not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated or account is inactive",
)
return current_user
# Role-based access control — dependency that checks user role
def require_role(required_role: str):
"""Return a dependency that checks the current user has at least the required role.
Usage: @router.get("/admin/endpoint", dependencies=[Depends(require_role("admin"))])
"""
async def role_checker(current_user: User = Depends(get_current_active_user)) -> User:
role_hierarchy = {"viewer": 1, "user": 2, "admin": 3}
user_level = role_hierarchy.get(current_user.role.value, 0)
required_level = role_hierarchy.get(required_role, 0)
if user_level < required_level:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Requires '{required_role}' role or higher",
)
return current_user
return role_checker
# Token refresh endpoint pattern
from fastapi import APIRouter, Request
auth_router = APIRouter(prefix="/auth", tags=["authentication"])
@auth_router.post("/token")
async def login_for_access_token(
request: Request,
) -> dict[str, str]:
"""OAuth2 password flow — exchange credentials for access + refresh tokens."""
form_data = await request.form()
username = form_data.get("username")
password = form_data.get("password")
if not username or not password:
raise HTTPException(status_code=400, detail="Username and password required")
# Authenticate user against database (simplified — real code queries DB)
user = await authenticate_user(username, password) # noqa: F821 — defined elsewhere
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(subject=str(user.id))
refresh_token = create_refresh_token(subject=str(user.id))
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer",
}
@auth_router.post("/refresh")
async def refresh_access_token(request: Request) -> dict[str, str]:
"""Refresh an expired access token using a valid refresh token."""
form_data = await request.form()
refresh_token = form_data.get("refresh_token")
if not refresh_token:
raise HTTPException(status_code=400, detail="Refresh token required")
try:
payload = decode_token(refresh_token)
if payload.get("type") != "refresh":
raise HTTPException(status_code=401, detail="Invalid token type")
subject = payload["sub"]
new_access_token = create_access_token(subject=subject)
return {"access_token": new_access_token, "token_type": "bearer"}
except HTTPException:
raise
except Exception:
raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
async def authenticate_user(username: str, password: str) -> User | None:
"""Authenticate user credentials against the database. Simplified for this example."""
# In real code: query DB, compare hashed passwords with verify_password()
return None # Placeholder
Pattern 5: Middleware Chain Construction
FastAPI uses Starlette's ASGI middleware infrastructure. Understand the difference between Starlette middleware (wrap each request) and ASGI middleware (wrap the entire application stack). Proper ordering matters — CORS must come before auth, timing before response generation.
# app/middleware/timing.py — Request timing middleware using Starlette's Middleware base class
import time
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import Response
class TimingMiddleware(BaseHTTPMiddleware):
"""Add X-Request-Time header with request processing duration in milliseconds.
Uses BaseHTTPMiddleware which wraps individual requests but runs within the ASGI scope.
Suitable for per-request logic like timing, logging, and basic auth checks.
"""
async def dispatch(
self,
request: Request,
call_next: RequestResponseEndpoint,
) -> Response:
start_time = time.perf_counter()
response = await call_next(request)
process_time = (time.perf_counter() - start_time) * 1000
# Add timing header to the response
response.headers["X-Request-Time"] = f"{process_time:.2f}ms"
# Log request details (in production, use structured JSON logging)
if process_time > 100: # Warn for slow requests (>100ms)
print(
f"[slow] {request.method} {request.url.path}
f"— {process_time:.2f}ms — {response.status_code}"
)
return response
# Alternative: ASGI middleware (runs at the application level, not per-request wrapped)
# This is more performant but has less request context access.
class AppLevelMiddleware:
"""ASGI-level middleware for high-performance concerns like gzip compression."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
# Check Accept-Encoding and set gzip header — applied before response body is generated
headers = dict(scope.get("headers", []))
accepts_gzip = b"application/gzip" in headers.get(b"accept-encoding", b"")
if accepts_gzip:
# In production, use a proper middleware like django-compressor or starlette's GZipMiddleware
pass # Placeholder — real implementation compresses response body
CORS configuration and custom exception handler middleware:
# app/core/exceptions.py — Global exception handlers for structured error responses
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from pydantic import ValidationError
class APIError(Exception):
"""Base application error with structured response payload."""
def __init__(self,
message: str,
status_code: int = status.HTTP_400_BAD_REQUEST,
detail: dict | None = None,
):
super().__init__(message)
self.message = message
self.status_code = status_code
self.detail = detail or {}
class NotFoundError(APIError):
def __init__(self, resource: str, identifier: str | None = None):
msg = f"{resource} not found"
if identifier:
msg += f": {identifier}"
super().__init__(
message=msg,
status_code=status.HTTP_404_NOT_FOUND,
detail={"resource": resource, "identifier": identifier},
)
class AuthenticationError(APIError):
def __init__(self, message: str = "Authentication required"):
super().__init__(
message=message,
status_code=status.HTTP_401_UNAUTHORIZED,
)
def register_exception_handlers(app: FastAPI) -> None:
"""Register all custom exception handlers on the FastAPI application.
These handlers intercept exceptions before they reach Starlette's default
error handler, producing consistent JSON error responses.
"""
@app.exception_handler(APIError)
async def api_error_handler(request: Request, exc: APIError) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content={
"error": {
"message": exc.message,
"status_code": exc.status_code,
"detail": exc.detail,
},
},
)
@app.exception_handler(ValidationError)
async def validation_error_handler(request: Request, exc: ValidationError) -> JSONResponse:
"""Convert Pydantic validation errors into structured API error responses.
This is critical — FastAPI automatically returns 422 for validation failures,
but this handler structures the response to match the API's error format.
"""
errors = []
for error in exc.errors():
loc = " -> ".join(str(l) for l in error.get("loc", ()))
errors.append({
"field": loc,
"message": error["msg"],
"type": error["type"],
})
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"error": {
"message": "Validation failed",
"status_code": status.HTTP_422_UNPROCESSABLE_ENTITY,
"errors": errors,
},
},
)
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
"""Catch-all handler for unexpected errors. Never expose internal details in production."""
# In production, log the full traceback internally but return a generic message externally
import logging
logger = logging.getLogger(__name__)
logger.exception("Unhandled exception in %s %s", request.method, request.url.path)
is_prod = request.headers.get("x-environment") == "production"
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
…(truncated)