fastapi-patterns

What this skill does

# FastAPI Patterns

## When to Use

Activate this skill when:
- Configuring FastAPI middleware (CORS, logging, timing, error handling)
- Creating complex dependency injection chains
- Implementing WebSocket endpoints with connection management
- Customizing OpenAPI documentation (tags, examples, deprecation)
- Setting up JWT authentication and role-based access dependencies
- Implementing background tasks (lightweight or distributed)
- Managing application lifecycle (startup/shutdown via lifespan)
- Setting up rate limiting or request throttling

Do NOT use this skill for:
- Basic endpoint CRUD, repository, or service patterns (use `python-backend-expert`)
- Writing tests for FastAPI endpoints (use `pytest-patterns`)
- API contract design or schema planning (use `api-design-patterns`)
- Architecture decisions (use `system-architecture`)

## Instructions

### Middleware Stack

Middleware executes in LIFO (Last In, First Out) order. The last middleware added is the outermost layer.

```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Order matters: added last = executed first (outermost)
app.add_middleware(TimingMiddleware)          # 3rd added = runs 1st
app.add_middleware(RequestLoggingMiddleware)  # 2nd added = runs 2nd
app.add_middleware(                           # 1st added = runs 3rd (innermost)
    CORSMiddleware,
    allow_origins=["https://app.example.com"],  # NEVER use "*" in production
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)
```

#### ASGI Middleware (Preferred)

Use pure ASGI middleware for performance-critical paths:

```python
from starlette.types import ASGIApp, Receive, Scope, Send
import time

class TimingMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start = time.perf_counter()
        await self.app(scope, receive, send)
        duration = time.perf_counter() - start
        # Log or record the duration
```

#### BaseHTTPMiddleware (Simpler but Slower)

Use only for middleware that needs to read/modify the request body or response:

```python
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response
```

**When to use which:**
- **ASGI middleware:** Performance-critical, no need to read request/response body
- **BaseHTTPMiddleware:** Need access to `Request`/`Response` objects, simpler API

### Authentication Dependencies

#### JWT Token Validation

```python
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

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

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: AsyncSession = Depends(get_async_session),
) -> User:
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

    user = await session.get(User, user_id)
    if user is None or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user
```

#### Role-Based Access (Factory Pattern)

```python
def require_role(*roles: str):
    """Factory that creates a dependency requiring specific roles."""
    async def check_role(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(
                status_code=403,
                detail=f"Requires one of: {', '.join(roles)}",
            )
        return user
    return check_role

# Usage in routes
@router.delete("/users/{user_id}", dependencies=[Depends(require_role("admin"))])
async def delete_user(user_id: int, ...) -> None:
    ...

@router.patch("/posts/{post_id}")
async def update_post(
    post_id: int,
    user: User = Depends(require_role("admin", "editor")),
) -> PostResponse:
    ...
```

### Dependency Injection Chains

#### Caching Behavior

FastAPI caches dependency results within a single request. The same dependency called multiple times returns the same instance:

```python
# get_async_session is called once per request, even if used by multiple deps
async def get_user_service(session: AsyncSession = Depends(get_async_session)) -> UserService:
    return UserService(session)

async def get_post_service(session: AsyncSession = Depends(get_async_session)) -> PostService:
    return PostService(session)  # Same session instance as user_service
```

To disable caching (get a new instance each time), use `use_cache=False`:
```python
session: AsyncSession = Depends(get_async_session, use_cache=False)
```

#### Yield Dependencies (Resource Cleanup)

```python
async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client
    # Client is automatically closed after the request
```

#### Overriding Dependencies in Tests

```python
# In tests
from app.main import app
from app.dependencies.auth import get_current_user

async def mock_current_user() -> User:
    return User(id=1, email="[email protected]", role="admin")

app.dependency_overrides[get_current_user] = mock_current_user
```

### Background Tasks

#### FastAPI BackgroundTasks (Lightweight)

For tasks that don't need to survive server restarts:

```python
from fastapi import BackgroundTasks

@router.post("/users", status_code=201)
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
    service: UserService = Depends(get_user_service),
) -> UserResponse:
    user = await service.create_user(data)
    background_tasks.add_task(send_welcome_email, user.email)
    return UserResponse.model_validate(user)

async def send_welcome_email(email: str) -> None:
    """Runs after the response is sent. Creates its own session."""
    async with async_session_factory() as session:
        async with session.begin():
            # Send email, log activity, etc.
            ...
```

**Rules:**
- Never reuse the request session in background tasks — create a new one
- Background tasks run in the same process — no retry, no persistence
- Use Celery or similar for tasks that need reliability, retry, or distribution

### WebSocket Pattern

```python
from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self) -> None:
        self.active: dict[int, list[WebSocket]] = {}

    async def connect(self, user_id: int, ws: WebSocket) -> None:
        await ws.accept()
        self.active.setdefault(user_id, []).append(ws)

    def disconnect(self, user_id: int, ws: WebSocket) -> None:
        if user_id in self.active:
            self.active[user_id].remove(ws)
            if not self.active[user_id]:
                del self.active[user_id]

    async def send_to_user(self, user_id: int, message: dict) -> None:
        for ws in self.active.get(user_id, []):
            await ws.send_json(message)

manager = ConnectionManager()

@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket, token: str) -> None:
    # Auth via query parameter: /ws?token=xxx
    user =