FastAPI Cheat Sheet: The Complete Quick Reference

The FastAPI patterns you need every day — routing, validation, dependency injection, auth, and testing — with copy-ready code in every section.

Quick reference

Project setup

pip install fastapi uvicorn[standard] pydantic-settings

Minimal app (main.py):

from fastapi import FastAPI

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

@app.get("/")
async def root():
    return {"message": "Hello World"}

uvicorn main:app --reload

Recommended project layout:

app/
├── main.py          # FastAPI instance, startup/shutdown
├── routers/
│   ├── users.py
│   └── items.py
├── models/          # SQLAlchemy models
├── schemas/         # Pydantic schemas
├── dependencies.py  # Shared dependencies
├── database.py      # DB session
└── config.py        # Settings

Routing

from fastapi import APIRouter

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

@router.get("/")
async def list_items():
    return []

@router.get("/{item_id}")
async def get_item(item_id: int):
    return {"id": item_id}

@router.post("/", status_code=201)
async def create_item(item: ItemCreate):
    return item

@router.put("/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    return item

@router.delete("/{item_id}", status_code=204)
async def delete_item(item_id: int):
    pass

Register in main.py:

from app.routers import items, users

app.include_router(items.router)
app.include_router(users.router, prefix="/api/v1")

Path and query parameters

from fastapi import Path, Query
from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"

# Path parameter with validation
@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(ge=1, le=1000, description="Item ID"),
):
    return {"id": item_id}

# Query parameters with defaults
@app.get("/items")
async def list_items(
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=10, ge=1, le=100),
    q: str | None = Query(default=None, max_length=50),
    active: bool = True,
):
    return {"skip": skip, "limit": limit, "q": q, "active": active}

# Enum path parameter
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    return {"model": model_name}

Request body with Pydantic

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)
    description: str | None = None
    tags: list[str] = []

class ItemResponse(ItemCreate):
    id: int
    created_at: datetime

    model_config = {"from_attributes": True}  # Pydantic v2 (ORM mode)

@app.post("/items", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate):
    # item is fully validated
    return {**item.model_dump(), "id": 1, "created_at": datetime.now()}

Nested models

class Address(BaseModel):
    street: str
    city: str
    country: str = "ME"

class User(BaseModel):
    name: str
    email: EmailStr          # pip install pydantic[email]
    address: Address
    tags: list[str] = []
    metadata: dict[str, str] = {}

Body + path + query together

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,           # path param
    item: ItemCreate,       # request body
    q: str | None = None,  # query param
):
    return {"id": item_id, "q": q, **item.model_dump()}

Response models and status codes

from fastapi import HTTPException
from fastapi.responses import JSONResponse, Response

# Restrict response fields
@app.get("/users/{user_id}", response_model=UserPublic)
async def get_user(user_id: int):
    user = db.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# Multiple response codes (for docs)
@app.post(
    "/items",
    response_model=ItemResponse,
    status_code=201,
    responses={
        400: {"description": "Invalid input"},
        409: {"description": "Item already exists"},
    },
)
async def create_item(item: ItemCreate): ...

# Custom headers
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return JSONResponse(
        content={"id": item_id},
        headers={"X-Request-ID": "abc123"},
    )

# Empty 204 response
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    return Response(status_code=204)

Dependency injection

from fastapi import Depends
from typing import Annotated

# Simple dependency
def get_db():
    db = SessionLocal()
    try:
        yield db          # yield → FastAPI calls cleanup after request
    finally:
        db.close()

DB = Annotated[Session, Depends(get_db)]

# Use in route
@app.get("/users")
async def list_users(db: DB):
    return db.query(User).all()

# Dependency with parameters
def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

Pagination = Annotated[dict, Depends(pagination)]

@app.get("/items")
async def list_items(page: Pagination):
    return page

# Nested dependencies
def verify_token(token: str = Header()):
    if token != "secret":
        raise HTTPException(status_code=401)
    return token

def get_current_user(token: str = Depends(verify_token), db: DB = Depends(get_db)):
    return db.query(User).filter_by(token=token).first()

CurrentUser = Annotated[User, Depends(get_current_user)]

# Router-level dependency (applies to all routes in router)
router = APIRouter(dependencies=[Depends(verify_token)])

Authentication

JWT Bearer auth

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt  # pip install PyJWT

SECRET_KEY = "your-secret-key"  # use env var in production
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def create_access_token(data: dict, expires_minutes: int = 30):
    payload = {**data, "exp": datetime.utcnow() + timedelta(minutes=expires_minutes)}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2_scheme), db: DB = Depends(get_db)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    user = db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return user

@app.post("/auth/token")
async def login(form: OAuth2PasswordRequestForm = Depends(), db: DB = Depends(get_db)):
    user = authenticate_user(db, form.username, form.password)
    if not user:
        raise HTTPException(status_code=401, detail="Bad credentials")
    token = create_access_token({"sub": str(user.id)})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/me")
async def me(current_user: User = Depends(get_current_user)):
    return current_user

API key auth

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="Invalid API key")
    return api_key

Request validation and error handling

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

# Custom validation error handler
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )

# Custom HTTP exception handler
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail},
    )

# Custom app exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc):
    return JSONResponse(status_code=404, content={"error": f"Item {exc.item_id} not found"})

Headers, cookies, and form data

from fastapi import Header, Cookie, Form, File, UploadFile

# Request headers
@app.get("/items")
async def list_items(
    user_agent: str | None = Header(default=None),
    x_request_id: str | None = Header(default=None),  # X-Request-Id → x_request_id
):
    return {"user_agent": user_agent}

# Cookies
@app.get("/profile")
async def get_profile(session_id: str | None = Cookie(default=None)):
    return {"session": session_id}

# Form data (application/x-www-form-urlencoded)
@app.post("/login")
async def login(username: str = Form(), password: str = Form()):
    return {"username": username}

# File upload (multipart/form-data)
@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    return {"filename": file.filename, "size": len(content), "type": file.content_type}

# Multiple files
@app.post("/upload-many")
async def upload_many(files: list[UploadFile]):
    return [{"filename": f.filename} for f in files]

Database with SQLAlchemy

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = "postgresql+psycopg2://user:password@localhost/dbname"

engine = create_engine(DATABASE_URL, pool_size=10, max_overflow=20)
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase):
    pass

# models.py
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy import ForeignKey

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    name: Mapped[str]
    is_active: Mapped[bool] = mapped_column(default=True)
    items: Mapped[list["Item"]] = relationship(back_populates="owner")

# Async SQLAlchemy (recommended for async FastAPI)
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

async_engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

async def get_async_db():
    async with AsyncSessionLocal() as session:
        yield session

Middleware

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Gzip compression
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Custom middleware (request timing)
@app.middleware("http")
async def add_process_time(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}s"
    return response

Background tasks

from fastapi import BackgroundTasks

def send_email(email: str, message: str):
    # runs after response is sent
    print(f"Sending email to {email}: {message}")

@app.post("/items")
async def create_item(item: ItemCreate, background_tasks: BackgroundTasks):
    # create item in DB...
    background_tasks.add_task(send_email, "user@example.com", f"Item {item.name} created")
    return item

Startup and shutdown events

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    await database.connect()
    print("Connected to database")
    yield
    # shutdown
    await database.disconnect()
    print("Database disconnected")

app = FastAPI(lifespan=lifespan)

Settings with pydantic-settings

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    app_name: str = "My API"
    debug: bool = False
    database_url: str
    secret_key: str
    allowed_origins: list[str] = ["http://localhost:3000"]

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()  # reads from .env automatically

# .env
# DATABASE_URL=postgresql+asyncpg://user:pass@localhost/mydb
# SECRET_KEY=super-secret-key

Testing with pytest

# pip install httpx pytest pytest-asyncio
from fastapi.testclient import TestClient
import pytest

from app.main import app

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_item():
    response = client.post(
        "/items",
        json={"name": "Widget", "price": 9.99},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Widget"

def test_validation_error():
    response = client.post("/items", json={"name": "", "price": -1})
    assert response.status_code == 422

# Override dependency in tests
from app.dependencies import get_db

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

# Async tests
@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        response = await ac.get("/")
    assert response.status_code == 200

Common patterns

Pagination

from dataclasses import dataclass

@dataclass
class Page:
    items: list
    total: int
    skip: int
    limit: int

    @property
    def has_more(self) -> bool:
        return self.skip + self.limit < self.total

@app.get("/items", response_model=PageResponse)
async def list_items(
    skip: int = Query(default=0, ge=0),
    limit: int = Query(default=20, ge=1, le=100),
    db: DB = Depends(get_db),
):
    total = db.query(Item).count()
    items = db.query(Item).offset(skip).limit(limit).all()
    return Page(items=items, total=total, skip=skip, limit=limit)

Versioned API

from fastapi import APIRouter

v1 = APIRouter(prefix="/api/v1")
v2 = APIRouter(prefix="/api/v2")

v1.include_router(items_v1.router, prefix="/items")
v2.include_router(items_v2.router, prefix="/items")

app.include_router(v1)
app.include_router(v2)

Rate limiting

# pip install slowapi
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("/items")
@limiter.limit("30/minute")
async def list_items(request: Request):  # request must be in signature
    return []

Common mistakes

FastAPI vs Django vs Flask

FAQ

Q: Should I use async def or def for route handlers? Use async def when your handler calls async functions (database, HTTP client, etc.). Use plain def only for CPU-bound work — FastAPI will run it in a thread pool. Mixing sync blocking calls inside async def will block the event loop.

Q: How do I handle CORS? Add CORSMiddleware before any routes. In development, set allow_origins=["*"]. In production, list specific origins. For credentialed requests (cookies, auth headers), set allow_credentials=True and list explicit origins — "*" won't work with credentials.

Q: What's the difference between response_model and the return type annotation? response_model controls what gets serialized and sent to the client — it filters and validates the output. The return type annotation is for IDE/type checker support only and doesn't affect the response. Always set response_model explicitly for public endpoints to avoid leaking fields.

Q: How do I connect FastAPI with a real database? Use SQLAlchemy with async session for best performance: create_async_engine + async_sessionmaker. Expose the session via Depends(get_async_db). For migrations, use Alembic (alembic init, alembic revision --autogenerate, alembic upgrade head).

Q: How do I deploy FastAPI to production? Run with uvicorn behind a reverse proxy (nginx or Caddy). Use gunicorn -k uvicorn.workers.UvicornWorker for multiple workers. On containers, set --workers $(nproc). For serverless: Vercel (via ASGI adapter), AWS Lambda (Mangum adapter), or Google Cloud Run (Docker container).

Q: How do I add background jobs (not just background tasks)? For simple fire-and-forget: BackgroundTasks. For reliable, persistent jobs: use Celery + Redis/RabbitMQ, or ARQ (async Redis Queue), or Dramatiq. Background tasks added via BackgroundTasks are lost if the process crashes.