FastAPI Reference: Routes, Pydantic, Auth, Async SQLAlchemy, Middleware & Testing

# pip install fastapi uvicorn[standard]

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, EmailStr, Field
from typing import Optional

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

# Pydantic model (request body + response):
class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    email: EmailStr                           # validates email format
    age: int = Field(gt=0, lt=150)            # greater than 0, less than 150
    role: str = "user"                        # default value

class User(BaseModel):
    id: int
    name: str
    email: EmailStr
    role: str

    model_config = {"from_attributes": True}  # allows ORM model → Pydantic (Pydantic v2)

# Path parameters:
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):             # int auto-validated from path
    user = db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# Query parameters + request body:
@app.post("/users", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate, send_welcome: bool = False):
    # user = validated UserCreate instance
    # send_welcome = ?send_welcome=true/false query param
    new_user = db.create_user(user.model_dump())
    if send_welcome:
        await send_email(new_user.email)
    return new_user

# Run locally:
# uvicorn main:app --reload
# API docs: http://localhost:8000/docs (Swagger UI)
# Alt docs: http://localhost:8000/redoc

from fastapi import Depends, Header
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt

# Database session dependency (shared across requests):
from sqlalchemy.orm import Session
from database import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db                              # provide db to endpoint
    finally:
        db.close()                            # cleanup after request

# Use in endpoint:
@app.get("/users/{id}")
async def get_user(id: int, db: Session = Depends(get_db)):
    return db.query(UserModel).filter(UserModel.id == id).first()

# JWT Bearer authentication:
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

# Login endpoint (exchange credentials for JWT):
@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form.username, form.password)
    if not user:
        raise HTTPException(status_code=401, detail="Wrong credentials")
    token = jwt.encode({"sub": str(user.id)}, SECRET_KEY, algorithm="HS256")
    return {"access_token": token, "token_type": "bearer"}

# Protected endpoint:
@app.get("/me", response_model=User)
async def get_me(current_user: User = Depends(get_current_user)):
    return current_user

from fastapi.middleware.cors import CORSMiddleware
from fastapi import BackgroundTasks
import time, logging

# CORS (required for browser-facing APIs):
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com", "http://localhost:3000"],
    allow_credentials=True,           # allows cookies/auth headers
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware (timing, logging):
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = f"{duration:.3f}s"
        logging.info(f"{request.method} {request.url.path} {response.status_code} {duration:.3f}s")
        return response

app.add_middleware(TimingMiddleware)

# Background tasks (fire-and-forget after response sent):
async def send_notification(user_id: int, message: str):
    await email_service.send(user_id, message)   # runs after response

@app.post("/orders")
async def create_order(order: OrderCreate, bg: BackgroundTasks):
    new_order = db.create_order(order.model_dump())
    bg.add_task(send_notification, new_order.user_id, "Order confirmed!")
    return new_order    # response sent immediately, notification sent after

# SQLModel (combines SQLAlchemy + Pydantic — from FastAPI's creator):
# pip install sqlmodel

from sqlmodel import SQLModel, Field, Session, create_engine, select
from typing import Optional

class UserDB(SQLModel, table=True):
    __tablename__ = "users"
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(max_length=100)
    email: str = Field(unique=True, index=True)
    active: bool = True

# Async SQLAlchemy 2.0 (for production async APIs):
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine("postgresql+asyncpg://user:pass@db:5432/mydb")

async def get_db() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(UserDB).where(UserDB.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(404, "Not found")
    return user

@app.post("/users", status_code=201)
async def create_user(data: UserCreate, db: AsyncSession = Depends(get_db)):
    user = UserDB(**data.model_dump())
    db.add(user)
    await db.commit()
    await db.refresh(user)     # reload from DB to get generated fields (id, timestamps)
    return user

# Alembic migrations:
# alembic init alembic
# alembic revision --autogenerate -m "add users table"
# alembic upgrade head

# Testing with httpx (async-native test client):
# pip install pytest pytest-asyncio httpx

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/users", json={
            "name": "Alice", "email": "alice@example.com", "age": 30
        })
    assert response.status_code == 201
    assert response.json()["name"] == "Alice"

@pytest.mark.asyncio
async def test_get_nonexistent_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/users/99999")
    assert response.status_code == 404

# Production Dockerfile (multi-stage, non-root, small image):
# FROM python:3.12-slim AS base
# RUN pip install uv
# COPY pyproject.toml .
# RUN uv pip install --system .
# COPY src/ /app/src/
# USER nobody
# CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# Performance tips:
# - Use async all the way down (async def endpoints + asyncpg/motor/redis.asyncio)
# - Workers = 2 * CPU_count + 1 (for I/O-bound): --workers $(expr 2 \* $(nproc) + 1)
# - Use response_model_exclude_unset=True to skip serializing None fields
# - Connection pooling: pool_size=10, max_overflow=20 in create_async_engine
# - Add Redis caching with aiocache or fastapi-cache2

# Health check endpoint (for K8s liveness/readiness probes):
@app.get("/health", include_in_schema=False)
async def health():
    return {"status": "ok"}