Python FastAPI service

Pydantic v2 models (validators and computed_fields)

# models/item.py — Pydantic v2 complete model definition
from pydantic import BaseModel, Field, field_validator, computed_field, model_validator
from decimal import Decimal
from typing import Optional
import re

class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200, description="Product name")
    price: Decimal = Field(gt=0, decimal_places=2, description="Price (positive)")
    sku: str = Field(description="Stock keeping unit, format XXX-000")
    discount_pct: Optional[float] = Field(default=None, ge=0, le=100)

    @field_validator('sku')
    @classmethod
    def validate_sku(cls, v: str) -> str:
        if not re.match(r'^[A-Z]{3}-\d{3}$', v):
            raise ValueError('SKU must match pattern XXX-000 (e.g. WGT-001)')
        return v

    @model_validator(mode='after')
    def check_discount_requires_price(self) -> 'ItemCreate':
        if self.discount_pct is not None and self.price < 10:
            raise ValueError('Discount only available for items priced >= 10')
        return self

class ItemResponse(ItemCreate):
    id: str
    created_at: str

    @computed_field  # type: ignore[misc]
    @property
    def final_price(self) -> Decimal:
        if self.discount_pct:
            return self.price * Decimal(1 - self.discount_pct / 100)
        return self.price

Request flow (Middleware → Depends → handler)

Typical resolution order for one HTTP call in FastAPI; exact middleware stack and handlers follow project registration order. Key point: dependencies resolve before the endpoint runs, and nested dependencies can be cached (same Depends per request may reuse the result).

  [ Client HTTP request ]
        │
        ▼
  ┌─────────────┐     CORS, trusted proxy, request_id, timeout (if custom)
  │  Middleware │
  └─────────────┘
        │
        ▼
  ┌─────────────┐     Match method + path; stack APIRouter prefixes
  │ Route match │
  └─────────────┘
        │
        ▼
  ┌─────────────┐     Depends in signature order (incl. sub-deps); may yield cleanup
  │ Depends chain│──── Common: Settings, DB session, CurrentUser, permissions
  └─────────────┘
        │
        ▼
  ┌─────────────┐     Query/Path/Body → Pydantic; return response_model
  │   Handler   │──── Uncaught → exception_handler / HTTPException
  └─────────────┘
        │
        ▼
  [ JSON / HTML / Stream / File / Redirect ]

Dependency injection: database / auth / permissions

# dependencies/database.py — DB session dependency
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from typing import AsyncGenerator, Annotated
from fastapi import Depends

engine = create_async_engine(settings.DATABASE_URL, pool_size=10, max_overflow=5)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# Type alias (recommended Annotated style)
DbSession = Annotated[AsyncSession, Depends(get_db)]

# dependencies/auth.py — JWT auth dependency
from fastapi import HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

bearer_scheme = HTTPBearer()

async def get_current_user(
    credentials: Annotated[HTTPAuthorizationCredentials, Depends(bearer_scheme)],
    db: DbSession,
) -> User:
    try:
        payload = jwt.decode(credentials.credentials, settings.JWT_SECRET, algorithms=['HS256'])
        user = await UserRepo(db).get_by_id(payload['sub'])
        if not user:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='User not found')
        return user
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail='Token expired')
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail='Invalid token')

# Permission check dependency (factory function)
def require_role(*roles: str):
    async def check(current_user: Annotated[User, Depends(get_current_user)]):
        if current_user.role not in roles:
            raise HTTPException(status_code=403, detail='Insufficient permissions')
        return current_user
    return check

CurrentUser = Annotated[User, Depends(get_current_user)]
AdminUser   = Annotated[User, Depends(require_role('admin'))]

# Route usage
@router.delete('/{item_id}')
async def delete_item(item_id: str, db: DbSession, _: AdminUser):
    ...

Async routes, background tasks, and TestClient

Async routes + BackgroundTasks:

# routers/orders.py — async routes + BackgroundTasks
from fastapi import APIRouter, BackgroundTasks
import asyncio

router = APIRouter(prefix='/orders', tags=['Orders'])

async def send_confirmation_email(order_id: str, email: str):
    """Background task: does not block the request response"""
    await asyncio.sleep(0)  # In production, integrate with email service
    print(f'Sending confirmation for order {order_id} to {email}')

@router.post('/', response_model=OrderResponse, status_code=201)
async def create_order(
    body: OrderCreate,
    background_tasks: BackgroundTasks,
    db: DbSession,
    current_user: CurrentUser,
):
    order = await OrderService(db).create(body, user_id=current_user.id)
    # Register background task (runs asynchronously after response is returned)
    background_tasks.add_task(send_confirmation_email, str(order.id), current_user.email)
    return order

# Middleware examples: request logging + gzip + CORS
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(CORSMiddleware,
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

# Request logging middleware
from starlette.middleware.base import BaseHTTPMiddleware
import time, uuid

class RequestLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        req_id = str(uuid.uuid4())
        request.state.req_id = req_id
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start
        print(f'{req_id} {request.method} {request.url.path} {response.status_code} {duration:.3f}s')
        response.headers['X-Request-Id'] = req_id
        return response

app.add_middleware(RequestLogMiddleware)

TestClient (with dependency_overrides):

# tests/test_orders.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock
from app.main import app
from app.dependencies.auth import get_current_user
from app.models.user import User

@pytest.fixture
def mock_user():
    return User(id='test-user-id', email='test@example.com', role='admin')

@pytest.fixture
def client(mock_user: User):
    # Override auth dependency — no real JWT needed
    app.dependency_overrides[get_current_user] = lambda: mock_user
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def test_create_order(client: TestClient):
    resp = client.post('/api/v1/orders', json={
        'items': [{'sku': 'WGT-001', 'quantity': 2}]
    })
    assert resp.status_code == 201
    data = resp.json()
    assert data['userId'] == 'test-user-id'
    assert 'id' in data

def test_create_order_validation_error(client: TestClient):
    resp = client.post('/api/v1/orders', json={'items': []})
    assert resp.status_code == 422
    assert resp.json()['detail'][0]['loc'] == ['body', 'items']

• Mount pools and clients on app.state or read them from dependency factories — single creation point.
• Create connection pools in lifespan; use dependency_overrides for in-memory fakes in tests.
• Align with ASGI server (Uvicorn) worker count: one connection pool per process.

Route prefix normalizer preview

Drafts often mix backslashes, omit a leading /, or add a trailing /. The tool below normalizes input for APIRouter(prefix=...) or include_router(..., prefix=...): single leading slash, collapse duplicate slashes, strip trailing slash (root stays /). This is a teaching convention—verify against your gateway and OpenAPI base URL before release.