Table of Contents
Quick Navigation
- FastAPI Architecture
- Async Patterns
- Dependency Injection
- Request Validation & Pydantic
- Authentication & Security
- Database Integration
- WebSockets & SSE
- Background Tasks
- Middleware & CORS
- Testing Strategies
- Performance Optimization
- Error Handling
- API Documentation
- Production Deployment
- Resources & Learning Path
FastAPI Architecture
Request Flow
┌─────────────────────────────────────────────────────────┐
│ Client Request │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ ASGI Server │
│ (Uvicorn, Hypercorn) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Middleware Stack │
│ ┌──────────────────────────────────────────────────┐ │
│ │ CORS Middleware │ │
│ │ Trusted Host Middleware │ │
│ │ HTTPS Redirect Middleware │ │
│ │ Custom Middleware │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Dependency Injection │
│ ┌──────────────────────────────────────────────────┐ │
│ │ • Database Session │ │
│ │ • Authentication │ │
│ │ • Configuration │ │
│ │ • Custom Dependencies │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Request Validation │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Pydantic Models │ │
│ │ • Path parameters │ │
│ │ • Query parameters │ │
│ │ • Request body │ │
│ │ • Headers & Cookies │ │
│ └──────────────────────────────────────────────────┘ │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Path Operation │
│ (Endpoint Handler) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Response Validation │
│ (Pydantic Response Model) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ JSON Serialization │
│ (orjson for performance) │
└──────────────────────┬──────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Client Response │
└─────────────────────────────────────────────────────────┘
Project Structure
myproject/ ├── app/ │ ├── __init__.py │ ├── main.py # App initialization │ ├── config.py # Settings │ │ │ ├── api/ # API routes │ │ ├── __init__.py │ │ ├── v1/ │ │ │ ├── __init__.py │ │ │ ├── endpoints/ │ │ │ │ ├── users.py │ │ │ │ ├── items.py │ │ │ │ └── auth.py │ │ │ └── router.py │ │ └── deps.py # Shared dependencies │ │ │ ├── core/ # Core functionality │ │ ├── __init__.py │ │ ├── security.py # Auth utilities │ │ ├── config.py # Configuration │ │ └── logging.py # Logging setup │ │ │ ├── models/ # SQLAlchemy models │ │ ├── __init__.py │ │ ├── user.py │ │ └── item.py │ │ │ ├── schemas/ # Pydantic schemas │ │ ├── __init__.py │ │ ├── user.py │ │ └── item.py │ │ │ ├── crud/ # Database operations │ │ ├── __init__.py │ │ ├── base.py │ │ ├── user.py │ │ └── item.py │ │ │ ├── db/ # Database │ │ ├── __init__.py │ │ ├── base.py │ │ ├── session.py │ │ └── init_db.py │ │ │ ├── services/ # Business logic │ │ ├── __init__.py │ │ ├── user_service.py │ │ └── email_service.py │ │ │ ├── middleware/ # Custom middleware │ │ ├── __init__.py │ │ └── timing.py │ │ │ └── tests/ # Tests │ ├── __init__.py │ ├── conftest.py │ ├── test_api/ │ └── test_services/ │ ├── alembic/ # Database migrations │ ├── versions/ │ └── env.py │ ├── scripts/ # Utility scripts │ ├── init_db.py │ └── seed_data.py │ ├── docker/ │ ├── Dockerfile │ └── docker-compose.yml │ ├── .env.example ├── requirements.txt ├── pyproject.toml ├── pytest.ini └── README.md Environment Variables (.env) PROJECT_NAME="My API" VERSION="1.0.0" API_V1_PREFIX="/api/v1" DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/db REDIS_URL=redis://localhost:6379/0 SECRET_KEY=your-secret-key-here ALGORITHM=HS256 ACCESS_TOKEN_EXPIRE_MINUTES=30 CORS_ORIGINS=["http://localhost:3000","https://yourdomain.com"] LOG_LEVEL=INFO ENVIRONMENT=production
Application Setup
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from app.core.config import settings
from app.api.v1.router import api_router
from app.middleware.timing import add_process_time_header
# Create FastAPI instance
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.VERSION,
openapi_url=f"{settings.API_V1_PREFIX}/openapi.json",
docs_url=f"{settings.API_V1_PREFIX}/docs",
redoc_url=f"{settings.API_V1_PREFIX}/redoc",
)
# Middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=settings.ALLOWED_HOSTS
)
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.middleware("http")(add_process_time_header)
# Include routers
app.include_router(
api_router,
prefix=settings.API_V1_PREFIX
)
@app.on_event("startup")
async def startup_event():
"""Initialize on startup"""
# Create database tables
# Initialize cache
# Start background tasks
pass
@app.on_event("shutdown")
async def shutdown_event():
"""Cleanup on shutdown"""
# Close database connections
# Stop background tasks
pass
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {"status": "healthy"}
# app/core/config.py
from pydantic_settings import BaseSettings
from typing import List, Optional
class Settings(BaseSettings):
# API
PROJECT_NAME: str = "FastAPI Project"
VERSION: str = "1.0.0"
API_V1_PREFIX: str = "/api/v1"
# Database
DATABASE_URL: str
# Redis
REDIS_URL: str
# Security
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# CORS
CORS_ORIGINS: List[str] = []
ALLOWED_HOSTS: List[str] = ["*"]
# Environment
ENVIRONMENT: str = "development"
LOG_LEVEL: str = "INFO"
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
# app/api/v1/router.py
from fastapi import APIRouter
from app.api.v1.endpoints import users, items, auth
api_router = APIRouter()
api_router.include_router(
auth.router,
prefix="/auth",
tags=["authentication"]
)
api_router.include_router(
users.router,
prefix="/users",
tags=["users"]
)
api_router.include_router(
items.router,
prefix="/items",
tags=["items"]
)
Async Patterns
Async Path Operations
# Basic async endpoint
from fastapi import FastAPI
import asyncio
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
# Async database query
result = await db.fetch_one(
"SELECT * FROM items WHERE id = :id",
{"id": item_id}
)
return result
# Parallel async operations
@app.get("/dashboard")
async def get_dashboard():
# Run multiple queries in parallel
users_task = asyncio.create_task(get_user_count())
items_task = asyncio.create_task(get_item_count())
orders_task = asyncio.create_task(get_order_count())
users, items, orders = await asyncio.gather(
users_task,
items_task,
orders_task
)
return {
"users": users,
"items": items,
"orders": orders
}
# Async with external API
import httpx
@app.get("/external-data")
async def get_external_data():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()
# Multiple external API calls in parallel
@app.get("/aggregated-data")
async def get_aggregated_data():
async with httpx.AsyncClient() as client:
api1_task = client.get("https://api1.example.com/data")
api2_task = client.get("https://api2.example.com/data")
api3_task = client.get("https://api3.example.com/data")
responses = await asyncio.gather(
api1_task,
api2_task,
api3_task,
return_exceptions=True
)
results = []
for resp in responses:
if isinstance(resp, Exception):
results.append({"error": str(resp)})
else:
results.append(resp.json())
return {"data": results}
# Async with timeout
@app.get("/data-with-timeout")
async def get_data_with_timeout():
try:
result = await asyncio.wait_for(
slow_operation(),
timeout=5.0
)
return result
except asyncio.TimeoutError:
raise HTTPException(
status_code=504,
detail="Operation timed out"
)
# Async streaming response
from fastapi.responses import StreamingResponse
async def generate_data():
for i in range(100):
await asyncio.sleep(0.1)
yield f"data: {i}\n\n"
@app.get("/stream")
async def stream_data():
return StreamingResponse(
generate_data(),
media_type="text/event-stream"
)
# Sync endpoint (when needed)
@app.get("/sync-operation")
def sync_endpoint():
# CPU-bound operation
result = complex_calculation()
return {"result": result}
# Mix sync in async (run in thread pool)
from fastapi.concurrency import run_in_threadpool
@app.get("/cpu-intensive")
async def cpu_intensive_endpoint():
# Run blocking operation in thread pool
result = await run_in_threadpool(
blocking_cpu_operation,
arg1,
arg2
)
return {"result": result}
# Async context manager
from contextlib import asynccontextmanager
@asynccontextmanager
async def get_database():
db = Database()
try:
await db.connect()
yield db
finally:
await db.disconnect()
@app.get("/data")
async def get_data():
async with get_database() as db:
result = await db.fetch("SELECT * FROM items")
return result
# Async generator for pagination
async def fetch_items_paginated(page_size: int = 100):
offset = 0
while True:
items = await db.fetch_many(
"SELECT * FROM items LIMIT :limit OFFSET :offset",
{"limit": page_size, "offset": offset}
)
if not items:
break
for item in items:
yield item
offset += page_size
@app.get("/all-items")
async def get_all_items():
items = []
async for item in fetch_items_paginated():
items.append(item)
return {"items": items}
# Error handling in async
@app.get("/items/{item_id}")
async def get_item(item_id: int):
try:
item = await db.fetch_one(
"SELECT * FROM items WHERE id = :id",
{"id": item_id}
)
if not item:
raise HTTPException(
status_code=404,
detail="Item not found"
)
return item
except Exception as e:
logger.error(f"Error fetching item: {e}")
raise HTTPException(
status_code=500,
detail="Internal server error"
)
Async Database Patterns
# SQLAlchemy async with asyncpg
from sqlalchemy.ext.asyncio import (
create_async_engine,
AsyncSession,
async_sessionmaker
)
from sqlalchemy.orm import declarative_base
# Database URL
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
# Create async engine
engine = create_async_engine(
DATABASE_URL,
echo=True,
pool_size=20,
max_overflow=0,
pool_pre_ping=True,
pool_recycle=3600,
)
# Session factory
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
Base = declarative_base()
# Dependency for database session
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
# Model definition
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
name = Column(String)
created_at = Column(DateTime, default=datetime.utcnow)
# CRUD operations
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
class UserCRUD:
async def create(
self,
db: AsyncSession,
email: str,
name: str
) -> User:
user = User(email=email, name=name)
db.add(user)
await db.commit()
await db.refresh(user)
return user
async def get(
self,
db: AsyncSession,
user_id: int
) -> User | None:
result = await db.execute(
select(User).where(User.id == user_id)
)
return result.scalar_one_or_none()
async def get_by_email(
self,
db: AsyncSession,
email: str
) -> User | None:
result = await db.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def get_multi(
self,
db: AsyncSession,
skip: int = 0,
limit: int = 100
) -> list[User]:
result = await db.execute(
select(User).offset(skip).limit(limit)
)
return result.scalars().all()
async def update(
self,
db: AsyncSession,
user_id: int,
name: str
) -> User | None:
user = await self.get(db, user_id)
if user:
user.name = name
await db.commit()
await db.refresh(user)
return user
async def delete(
self,
db: AsyncSession,
user_id: int
) -> bool:
user = await self.get(db, user_id)
if user:
await db.delete(user)
await db.commit()
return True
return False
user_crud = UserCRUD()
# Using in endpoint
from fastapi import Depends
@app.post("/users", response_model=UserSchema)
async def create_user(
user: UserCreate,
db: AsyncSession = Depends(get_db)
):
# Check if user exists
existing = await user_crud.get_by_email(db, user.email)
if existing:
raise HTTPException(
status_code=400,
detail="Email already registered"
)
# Create user
new_user = await user_crud.create(
db,
email=user.email,
name=user.name
)
return new_user
@app.get("/users/{user_id}", response_model=UserSchema)
async def read_user(
user_id: int,
db: AsyncSession = Depends(get_db)
):
user = await user_crud.get(db, user_id)
if not user:
raise HTTPException(
status_code=404,
detail="User not found"
)
return user
# Async with Redis
import aioredis
from typing import Optional
class RedisCache:
def __init__(self, redis_url: str):
self.redis_url = redis_url
self.redis: Optional[aioredis.Redis] = None
async def connect(self):
self.redis = await aioredis.from_url(
self.redis_url,
encoding="utf-8",
decode_responses=True
)
async def disconnect(self):
if self.redis:
await self.redis.close()
async def get(self, key: str) -> Optional[str]:
return await self.redis.get(key)
async def set(
self,
key: str,
value: str,
expire: int = 300
):
await self.redis.setex(key, expire, value)
async def delete(self, key: str):
await self.redis.delete(key)
cache = RedisCache(settings.REDIS_URL)
@app.on_event("startup")
async def startup():
await cache.connect()
@app.on_event("shutdown")
async def shutdown():
await cache.disconnect()
# Cache decorator
import json
from functools import wraps
def cached(expire: int = 300):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# Generate cache key
cache_key = f"{func.__name__}:{json.dumps(kwargs)}"
# Try to get from cache
cached_value = await cache.get(cache_key)
if cached_value:
return json.loads(cached_value)
# Execute function
result = await func(*args, **kwargs)
# Store in cache
await cache.set(
cache_key,
json.dumps(result),
expire
)
return result
return wrapper
return decorator
@app.get("/expensive-operation")
@cached(expire=600)
async def expensive_operation():
# Expensive computation
result = await compute_something()
return result
Dependency Injection
Basic Dependencies
# Simple dependency
from fastapi import Depends
async def get_current_user(token: str = Header(...)):
user = await verify_token(token)
if not user:
raise HTTPException(
status_code=401,
detail="Invalid authentication"
)
return user
@app.get("/users/me")
async def read_current_user(
current_user: User = Depends(get_current_user)
):
return current_user
# Database session dependency
from sqlalchemy.ext.asyncio import AsyncSession
async def get_db():
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
@app.post("/items")
async def create_item(
item: ItemCreate,
db: AsyncSession = Depends(get_db)
):
new_item = Item(**item.dict())
db.add(new_item)
await db.commit()
return new_item
# Query parameter dependencies
from typing import Optional
class CommonQueryParams:
def __init__(
self,
skip: int = 0,
limit: int = 100,
sort: Optional[str] = None
):
self.skip = skip
self.limit = limit
self.sort = sort
@app.get("/items")
async def read_items(
commons: CommonQueryParams = Depends()
):
items = await get_items(
skip=commons.skip,
limit=commons.limit,
sort=commons.sort
)
return items
# Sub-dependencies
async def get_token(token: str = Header(...)):
return token
async def get_current_user(token: str = Depends(get_token)):
user = await verify_token(token)
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user)
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
@app.get("/users/me")
async def read_user_me(
current_user: User = Depends(get_current_active_user)
):
return current_user
# Class-based dependencies
class PermissionChecker:
def __init__(self, required_permission: str):
self.required_permission = required_permission
async def __call__(
self,
current_user: User = Depends(get_current_user)
):
if self.required_permission not in current_user.permissions:
raise HTTPException(
status_code=403,
detail="Not enough permissions"
)
return current_user
require_admin = PermissionChecker("admin")
@app.delete("/items/{item_id}")
async def delete_item(
item_id: int,
current_user: User = Depends(require_admin)
):
await delete_item_by_id(item_id)
return {"status": "deleted"}
# Global dependencies
app = FastAPI(
dependencies=[Depends(verify_api_key)]
)
# Or for router
router = APIRouter(
prefix="/admin",
tags=["admin"],
dependencies=[Depends(require_admin)]
)
# Dependency with yield
async def get_db_connection():
connection = await create_connection()
try:
yield connection
finally:
await connection.close()
@app.get("/data")
async def get_data(
conn = Depends(get_db_connection)
):
result = await conn.fetch("SELECT * FROM items")
return result
# Caching dependencies
from functools import lru_cache
@lru_cache()
def get_settings():
return Settings()
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
return {"app_name": settings.PROJECT_NAME}
# Optional dependencies
async def get_optional_user(
token: Optional[str] = Header(None)
) -> Optional[User]:
if token:
return await verify_token(token)
return None
@app.get("/items")
async def read_items(
user: Optional[User] = Depends(get_optional_user)
):
if user:
# Return user-specific items
return await get_user_items(user.id)
# Return public items
return await get_public_items()
# Dependencies in path operations
@app.get(
"/items",
dependencies=[
Depends(verify_api_key),
Depends(rate_limiter)
]
)
async def read_items():
return await get_items()
# Advanced: Dependency override for testing
async def override_get_db():
# Test database
pass
app.dependency_overrides[get_db] = override_get_db
Advanced Dependency Patterns
# Factory pattern dependency
def create_service_dependency(service_class):
async def get_service(
db: AsyncSession = Depends(get_db)
):
return service_class(db)
return get_service
class UserService:
def __init__(self, db: AsyncSession):
self.db = db
async def create_user(self, email: str, name: str):
user = User(email=email, name=name)
self.db.add(user)
await self.db.commit()
return user
get_user_service = create_service_dependency(UserService)
@app.post("/users")
async def create_user(
user_data: UserCreate,
service: UserService = Depends(get_user_service)
):
return await service.create_user(
user_data.email,
user_data.name
)
# Repository pattern
class BaseRepository:
def __init__(self, db: AsyncSession):
self.db = db
class UserRepository(BaseRepository):
async def get_by_id(self, user_id: int) -> Optional[User]:
result = await self.db.execute(
select(User).where(User.id == user_id)
)
return result.scalar_one_or_none()
async def get_by_email(self, email: str) -> Optional[User]:
result = await self.db.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def create(self, user: User) -> User:
self.db.add(user)
await self.db.commit()
await self.db.refresh(user)
return user
async def get_user_repo(
db: AsyncSession = Depends(get_db)
) -> UserRepository:
return UserRepository(db)
@app.get("/users/{user_id}")
async def read_user(
user_id: int,
repo: UserRepository = Depends(get_user_repo)
):
user = await repo.get_by_id(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
# Service layer pattern
class UserService:
def __init__(
self,
repo: UserRepository,
cache: RedisCache,
email_service: EmailService
):
self.repo = repo
self.cache = cache
self.email_service = email_service
async def create_user(
self,
email: str,
name: str
) -> User:
# Check if exists
existing = await self.repo.get_by_email(email)
if existing:
raise ValueError("Email already registered")
# Create user
user = User(email=email, name=name)
user = await self.repo.create(user)
# Send welcome email
await self.email_service.send_welcome(user.email)
# Clear cache
await self.cache.delete(f"user:{user.id}")
return user
async def get_user_service(
repo: UserRepository = Depends(get_user_repo),
cache: RedisCache = Depends(get_cache),
email_service: EmailService = Depends(get_email_service)
) -> UserService:
return UserService(repo, cache, email_service)
# Unit of Work pattern
class UnitOfWork:
def __init__(self, session: AsyncSession):
self.session = session
self.users = UserRepository(session)
self.items = ItemRepository(session)
async def commit(self):
await self.session.commit()
async def rollback(self):
await self.session.rollback()
async def get_uow(
db: AsyncSession = Depends(get_db)
) -> UnitOfWork:
return UnitOfWork(db)
@app.post("/users-with-items")
async def create_user_with_items(
data: UserWithItemsCreate,
uow: UnitOfWork = Depends(get_uow)
):
try:
# Create user
user = await uow.users.create(
User(email=data.email, name=data.name)
)
# Create items
for item_data in data.items:
item = Item(**item_data.dict(), user_id=user.id)
await uow.items.create(item)
# Commit all changes
await uow.commit()
return user
except Exception:
await uow.rollback()
raise
# Dependency with context
from contextvars import ContextVar
request_id_var: ContextVar[str] = ContextVar('request_id')
async def get_request_id() -> str:
return request_id_var.get()
@app.middleware("http")
async def add_request_id(request: Request, call_next):
request_id = str(uuid.uuid4())
request_id_var.set(request_id)
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
@app.get("/test")
async def test_endpoint(
request_id: str = Depends(get_request_id)
):
return {"request_id": request_id}
Request Validation & Pydantic
Pydantic Models
# Basic schema
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime
class UserBase(BaseModel):
email: EmailStr
name: str = Field(..., min_length=1, max_length=100)
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=100)
email: Optional[EmailStr] = None
class UserInDB(UserBase):
id: int
created_at: datetime
is_active: bool = True
class Config:
from_attributes = True # Was orm_mode in Pydantic v1
class UserResponse(UserBase):
id: int
created_at: datetime
# Advanced validation
from pydantic import validator, root_validator
class UserCreate(BaseModel):
email: EmailStr
password: str
password_confirm: str
age: int
@validator('password')
def validate_password(cls, v):
if len(v) < 8:
raise ValueError('Password must be at least 8 characters')
if not any(c.isupper() for c in v):
raise ValueError('Password must contain uppercase')
if not any(c.isdigit() for c in v):
raise ValueError('Password must contain digit')
return v
@validator('age')
def validate_age(cls, v):
if v < 18:
raise ValueError('Must be at least 18 years old')
if v > 120:
raise ValueError('Age must be realistic')
return v
@root_validator
def check_passwords_match(cls, values):
pw = values.get('password')
pw_confirm = values.get('password_confirm')
if pw != pw_confirm:
raise ValueError('Passwords do not match')
return values
# Field validation
from pydantic import Field, constr, conint, confloat
class Product(BaseModel):
name: constr(min_length=1, max_length=100)
price: confloat(gt=0, le=1000000)
quantity: conint(ge=0)
sku: constr(regex=r'^[A-Z]{3}-\d{6}$')
description: Optional[str] = Field(None, max_length=1000)
# Alternative using Field
weight: float = Field(..., gt=0, le=1000, description="Weight in kg")
tags: list[str] = Field(default_factory=list, max_items=10)
# Nested models
class Address(BaseModel):
street: str
city: str
state: str
zip_code: str = Field(..., regex=r'^\d{5}$')
country: str = "USA"
class User(BaseModel):
name: str
email: EmailStr
address: Address
shipping_addresses: list[Address] = []
# Union types
from typing import Union
class Dog(BaseModel):
type: str = "dog"
breed: str
class Cat(BaseModel):
type: str = "cat"
color: str
class Pet(BaseModel):
name: str
animal: Union[Dog, Cat]
# Discriminated unions
from typing import Literal
class Dog(BaseModel):
pet_type: Literal["dog"]
breed: str
class Cat(BaseModel):
pet_type: Literal["cat"]
color: str
class PetOwner(BaseModel):
name: str
pet: Union[Dog, Cat] = Field(..., discriminator='pet_type')
# Custom validators
from pydantic import validator
class BookCreate(BaseModel):
title: str
isbn: str
@validator('isbn')
def validate_isbn(cls, v):
# Remove hyphens
v = v.replace('-', '')
if len(v) != 13:
raise ValueError('ISBN must be 13 digits')
if not v.isdigit():
raise ValueError('ISBN must contain only digits')
return v
# Pre and post validators
class User(BaseModel):
name: str
email: str
@validator('name', pre=True)
def name_must_not_be_empty(cls, v):
if isinstance(v, str):
v = v.strip()
if not v:
raise ValueError('Name cannot be empty')
return v
@validator('email')
def email_must_be_lowercase(cls, v):
return v.lower()
# Config options
class User(BaseModel):
id: int
name: str
email: EmailStr
class Config:
from_attributes = True # Enable ORM mode
validate_assignment = True # Validate on assignment
arbitrary_types_allowed = True # Allow arbitrary types
use_enum_values = True # Use enum values instead of objects
json_encoders = {
datetime: lambda v: v.isoformat()
}
# Generic models
from typing import TypeVar, Generic
T = TypeVar('T')
class Page(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
size: int
pages: int
class UserPage(Page[UserResponse]):
pass
@app.get("/users", response_model=Page[UserResponse])
async def list_users(page: int = 1, size: int = 20):
users = await get_users(page, size)
total = await count_users()
return Page(
items=users,
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)
Request Validation Patterns
# Path parameters
@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(..., gt=0, le=1000)
):
return {"item_id": item_id}
# Query parameters
from typing import Optional
@app.get("/items")
async def list_items(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100),
search: Optional[str] = Query(None, min_length=3),
sort: str = Query("created_at", regex="^(name|created_at|price)$")
):
return await get_items(skip, limit, search, sort)
# Request body
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
description: Optional[str] = None
@app.post("/items")
async def create_item(item: Item):
return item
# Multiple body parameters
@app.put("/items/{item_id}")
async def update_item(
item_id: int,
item: Item,
user: User,
importance: int = Body(...)
):
return {
"item_id": item_id,
"item": item,
"user": user,
"importance": importance
}
# Body with embed
@app.post("/items")
async def create_item(
item: Item = Body(..., embed=True)
):
# Expects: {"item": {"name": "...", "price": ...}}
return item
# Form data
from fastapi import Form
@app.post("/login")
async def login(
username: str = Form(...),
password: str = Form(...)
):
return {"username": username}
# File upload
from fastapi import File, UploadFile
@app.post("/upload")
async def upload_file(
file: UploadFile = File(...)
):
contents = await file.read()
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(contents)
}
# Multiple file upload
@app.post("/uploadfiles")
async def upload_files(
files: list[UploadFile] = File(...)
):
return {
"filenames": [file.filename for file in files]
}
# File with form data
@app.post("/upload-with-data")
async def upload_with_data(
file: UploadFile = File(...),
description: str = Form(...),
tags: list[str] = Form([])
):
return {
"filename": file.filename,
"description": description,
"tags": tags
}
# Headers
from fastapi import Header
@app.get("/items")
async def read_items(
user_agent: Optional[str] = Header(None),
x_request_id: str = Header(..., alias="X-Request-ID")
):
return {
"User-Agent": user_agent,
"X-Request-ID": x_request_id
}
# Cookies
from fastapi import Cookie
@app.get("/items")
async def read_items(
session_id: Optional[str] = Cookie(None)
):
return {"session_id": session_id}
# Response model
class UserResponse(BaseModel):
id: int
email: str
name: str
class Config:
from_attributes = True
@app.get("/users/{user_id}", response_model=UserResponse)
async def read_user(user_id: int):
user = await get_user(user_id)
return user # Automatically validated and serialized
# Response model with exclude
@app.get(
"/users/{user_id}",
response_model=User,
response_model_exclude={"password", "internal_id"}
)
async def read_user(user_id: int):
return await get_user(user_id)
# Response model with include
@app.get(
"/users/{user_id}",
response_model=User,
response_model_include={"id", "email", "name"}
)
async def read_user(user_id: int):
return await get_user(user_id)
# Multiple response models
from typing import Union
@app.get(
"/items/{item_id}",
response_model=Union[Item, ErrorResponse]
)
async def read_item(item_id: int):
item = await get_item(item_id)
if not item:
return ErrorResponse(error="Item not found")
return item
# List response
@app.get("/users", response_model=list[UserResponse])
async def list_users():
users = await get_all_users()
return users
# Custom response class
from fastapi.responses import JSONResponse
@app.get("/custom")
async def custom_response():
return JSONResponse(
content={"message": "Hello"},
headers={"X-Custom-Header": "Value"}
)
Authentication & Security
JWT Authentication
# JWT utilities
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
# JWT token creation
SECRET_KEY = "your-secret-key-here"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def create_access_token(
data: dict,
expires_delta: Optional[timedelta] = None
):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def decode_access_token(token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
# Authentication schemas
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
email: Optional[str] = None
class UserLogin(BaseModel):
email: EmailStr
password: str
# Get current user dependency
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_access_token(token)
if payload is None:
raise credentials_exception
email: str = payload.get("sub")
if email is None:
raise credentials_exception
user = await get_user_by_email(db, email)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user)
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
# Login endpoint
from fastapi.security import OAuth2PasswordRequestForm
@app.post("/token", response_model=Token)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db)
):
# Authenticate user
user = await get_user_by_email(db, form_data.username)
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=401,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
# Create access token
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email},
expires_delta=access_token_expires
)
return {
"access_token": access_token,
"token_type": "bearer"
}
# Register endpoint
@app.post("/register", response_model=UserResponse)
async def register(
user_data: UserCreate,
db: AsyncSession = Depends(get_db)
):
# Check if user exists
existing = await get_user_by_email(db, user_data.email)
if existing:
raise HTTPException(
status_code=400,
detail="Email already registered"
)
# Create user
hashed_password = get_password_hash(user_data.password)
user = User(
email=user_data.email,
name=user_data.name,
hashed_password=hashed_password
)
db.add(user)
await db.commit()
await db.refresh(user)
return user
# Protected endpoint
@app.get("/users/me", response_model=UserResponse)
async def read_users_me(
current_user: User = Depends(get_current_active_user)
):
return current_user
# Refresh token pattern
def create_refresh_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=7)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
@app.post("/token/refresh")
async def refresh_token(
refresh_token: str,
db: AsyncSession = Depends(get_db)
):
payload = decode_access_token(refresh_token)
if not payload:
raise HTTPException(status_code=401, detail="Invalid token")
email = payload.get("sub")
user = await get_user_by_email(db, email)
if not user:
raise HTTPException(status_code=401, detail="User not found")
# Create new access token
access_token = create_access_token(data={"sub": user.email})
return {
"access_token": access_token,
"token_type": "bearer"
}
Security Best Practices
# API Key authentication
from fastapi.security import APIKeyHeader
API_KEY_HEADER = APIKeyHeader(name="X-API-Key")
async def verify_api_key(api_key: str = Depends(API_KEY_HEADER)):
if api_key != settings.API_KEY:
raise HTTPException(
status_code=403,
detail="Could not validate API key"
)
return api_key
@app.get("/admin/stats")
async def get_stats(
api_key: str = Depends(verify_api_key)
):
return {"stats": "data"}
# Rate limiting
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.get("/limited")
@limiter.limit("5/minute")
async def limited_endpoint(request: Request):
return {"message": "This endpoint is rate limited"}
# CORS configuration
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:3000",
"https://yourdomain.com"
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["X-Total-Count"],
max_age=3600,
)
# HTTPS redirect
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
if settings.ENVIRONMENT == "production":
app.add_middleware(HTTPSRedirectMiddleware)
# Trusted hosts
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=[
"yourdomain.com",
"*.yourdomain.com"
]
)
# Security headers middleware
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
# Input sanitization
import bleach
def sanitize_html(html: str) -> str:
allowed_tags = ['p', 'br', 'strong', 'em', 'a']
allowed_attrs = {'a': ['href', 'title']}
return bleach.clean(
html,
tags=allowed_tags,
attributes=allowed_attrs,
strip=True
)
# SQL injection prevention (use ORMs)
# ✅ Good: Using SQLAlchemy
result = await db.execute(
select(User).where(User.email == user_input)
)
# ❌ Bad: Raw SQL with string formatting
# query = f"SELECT * FROM users WHERE email = '{user_input}'"
# Password validation
import re
def validate_password_strength(password: str) -> bool:
if len(password) < 8:
return False
if not re.search(r'[A-Z]', password):
return False
if not re.search(r'[a-z]', password):
return False
if not re.search(r'\d', password):
return False
if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
return False
return True
# File upload validation
ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.pdf'}
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB
async def validate_file(file: UploadFile):
# Check extension
ext = os.path.splitext(file.filename)[1].lower()
if ext not in ALLOWED_EXTENSIONS:
raise HTTPException(
status_code=400,
detail=f"File type {ext} not allowed"
)
# Check file size
contents = await file.read()
if len(contents) > MAX_FILE_SIZE:
raise HTTPException(
status_code=400,
detail="File too large"
)
# Reset file pointer
await file.seek(0)
return file
# CSRF protection for cookies
from starlette.middleware.sessions import SessionMiddleware
app.add_middleware(
SessionMiddleware,
secret_key=settings.SECRET_KEY,
session_cookie="session",
max_age=3600,
same_site="lax",
https_only=True
)
Database Integration
📚 Content Note
This section covers async database patterns with SQLAlchemy, connection pooling, migrations with Alembic, and query optimization. See earlier "Async Database Patterns" section for detailed examples. Key Topics: • SQLAlchemy async with asyncpg/aiomysql • Session management and connection pooling • Alembic migrations • Query optimization and indexing • Transaction management • Database testing strategies Related Sheets: → PostgreSQL Production → SQLAlchemy Advanced → Database Design Patterns
Resources & Learning Path
Learning Progression
Phase 1: FastAPI Basics (1-2 weeks) □ Path operations and routing □ Request/response models with Pydantic □ Query and path parameters □ Request body validation □ Response models □ Basic error handling Phase 2: Intermediate FastAPI (2-3 weeks) □ Dependency injection □ Database integration (SQLAlchemy) □ Authentication (JWT) □ File uploads □ Background tasks □ Middleware basics Phase 3: Advanced FastAPI (3-4 weeks) □ Async patterns and optimization □ WebSockets □ Advanced dependency patterns □ Testing with pytest □ API documentation customization □ Performance optimization Phase 4: Production FastAPI (Ongoing) □ Deployment strategies □ Monitoring and logging □ Security hardening □ Microservices architecture □ GraphQL integration □ Event-driven patterns
Related Comprehensive Sheets
Python Ecosystem → Python Advanced (async patterns, performance) → Django Production (alternative framework) → Flask Production (microservices) → Pydantic Advanced (validation) → SQLAlchemy (ORM patterns) API Development → REST API Design → GraphQL with FastAPI → WebSockets Advanced → API Security Patterns → OpenAPI/Swagger Testing & Quality → Pytest Advanced → API Testing Strategies → Load Testing → Integration Testing Deployment & DevOps → Docker & Kubernetes → CI/CD Pipelines → Monitoring & Observability → AWS/GCP Deployment → Nginx Configuration
Pro Tips Summary
Performance ✓ Use async/await for I/O operations ✓ Implement connection pooling ✓ Use response_model to filter data ✓ Enable GZip compression ✓ Cache expensive operations ✓ Use uvloop for better performance ✓ Optimize database queries Security ✓ Always use HTTPS in production ✓ Implement proper authentication ✓ Validate all input data ✓ Use environment variables for secrets ✓ Enable CORS selectively ✓ Implement rate limiting ✓ Keep dependencies updated Code Quality ✓ Use Pydantic models for validation ✓ Leverage dependency injection ✓ Write comprehensive tests ✓ Document your API ✓ Use type hints everywhere ✓ Follow RESTful conventions ✓ Handle errors gracefully Development ✓ Use FastAPI's auto docs ✓ Test with pytest-asyncio ✓ Use Alembic for migrations ✓ Implement proper logging ✓ Monitor application metrics ✓ Use background tasks wisely ✓ Profile and optimize bottlenecks