"""
Production-ready FastAPI application template.

Usage:
    uvicorn main:app --reload  # Development
    uvicorn main:app --host 0.0.0.0 --port 8000  # Production
"""

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings


# =============================================================================
# Configuration
# =============================================================================

class Settings(BaseSettings):
    """Application settings from environment variables."""

    app_name: str = "My API"
    debug: bool = False
    database_url: str = "postgresql+asyncpg://user:pass@localhost/db"
    redis_url: str = "redis://localhost:6379/0"
    api_key: str = ""

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}


# Cache settings
from functools import lru_cache

@lru_cache
def get_settings() -> Settings:
    return Settings()


# =============================================================================
# Lifespan Management
# =============================================================================

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application startup and shutdown."""
    settings = get_settings()

    # Startup
    # app.state.db = await create_db_pool(settings.database_url)
    # app.state.redis = await create_redis_client(settings.redis_url)
    print(f"Starting {settings.app_name}...")

    yield

    # Shutdown
    # await app.state.db.close()
    # await app.state.redis.close()
    print("Shutting down...")


# =============================================================================
# Application
# =============================================================================

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

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"] if get_settings().debug else ["https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# =============================================================================
# Error Handling
# =============================================================================

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Handle unhandled exceptions."""
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )


# =============================================================================
# Dependencies
# =============================================================================

async def get_db():
    """Database session dependency."""
    # async with async_session() as session:
    #     yield session
    yield None  # Placeholder


DB = Annotated[None, Depends(get_db)]  # Replace None with actual type


# =============================================================================
# Models
# =============================================================================

class HealthResponse(BaseModel):
    status: str
    version: str


class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str | None = None


class ItemResponse(BaseModel):
    id: int
    name: str
    description: str | None

    model_config = {"from_attributes": True}


# =============================================================================
# Routes
# =============================================================================

@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check endpoint."""
    return HealthResponse(status="healthy", version="1.0.0")


@app.get("/items", response_model=list[ItemResponse])
async def list_items(
    db: DB,
    skip: int = 0,
    limit: int = 10,
):
    """List all items."""
    # items = await db.execute(select(Item).offset(skip).limit(limit))
    # return items.scalars().all()
    return []


@app.post("/items", response_model=ItemResponse, status_code=201)
async def create_item(item: ItemCreate, db: DB):
    """Create a new item."""
    # db_item = Item(**item.model_dump())
    # db.add(db_item)
    # await db.commit()
    # await db.refresh(db_item)
    # return db_item
    return ItemResponse(id=1, name=item.name, description=item.description)


@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int, db: DB):
    """Get a single item."""
    # item = await db.get(Item, item_id)
    # if not item:
    #     raise HTTPException(status_code=404, detail="Item not found")
    # return item
    raise HTTPException(status_code=404, detail="Item not found")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

# FastAPI Background Tasks

Async background processing patterns.

## Built-in BackgroundTasks

```python
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

async def send_email(email: str, message: str):
    """Background task - runs after response sent."""
    # Simulate email sending
    await asyncio.sleep(2)
    print(f"Email sent to {email}: {message}")

@app.post("/signup")
async def signup(
    email: str,
    background_tasks: BackgroundTasks,
):
    # Create user synchronously
    user = create_user(email)

    # Queue background task
    background_tasks.add_task(send_email, email, "Welcome!")

    # Response sent immediately
    return {"message": "User created"}


# Multiple tasks
@app.post("/order")
async def create_order(
    order: OrderCreate,
    background_tasks: BackgroundTasks,
):
    db_order = save_order(order)

    # Queue multiple tasks
    background_tasks.add_task(send_confirmation, order.email)
    background_tasks.add_task(update_inventory, order.items)
    background_tasks.add_task(notify_warehouse, db_order.id)

    return {"order_id": db_order.id}
```

## Dependency Injection with Background Tasks

```python
from fastapi import Depends, BackgroundTasks
from typing import Annotated

async def audit_log(action: str, user_id: int):
    """Log user actions."""
    await db.execute(
        "INSERT INTO audit_log (action, user_id) VALUES ($1, $2)",
        action, user_id
    )

def get_auditor(background_tasks: BackgroundTasks):
    """Factory for audit logging."""
    def log(action: str, user_id: int):
        background_tasks.add_task(audit_log, action, user_id)
    return log

Auditor = Annotated[Callable, Depends(get_auditor)]

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: CurrentUser,
    auditor: Auditor,
):
    await db.delete_user(user_id)
    auditor("user_deleted", current_user.id)
    return {"deleted": user_id}
```

## Longer Tasks with Celery

```python
# tasks.py
from celery import Celery

celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0",
)

@celery_app.task
def process_video(video_id: int):
    """Long-running task - handled by Celery worker."""
    video = get_video(video_id)
    processed = transcode(video)
    save_processed(processed)
    return {"status": "completed", "video_id": video_id}


# api.py
from fastapi import FastAPI
from tasks import process_video

app = FastAPI()

@app.post("/videos/{video_id}/process")
async def start_processing(video_id: int):
    # Queue task in Celery
    task = process_video.delay(video_id)

    return {
        "task_id": task.id,
        "status": "queued",
    }

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    task = process_video.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None,
    }
```

## Periodic Tasks with APScheduler

```python
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from fastapi import FastAPI
from contextlib import asynccontextmanager

scheduler = AsyncIOScheduler()

async def cleanup_expired_sessions():
    """Run daily at midnight."""
    await db.execute("DELETE FROM sessions WHERE expires < NOW()")

async def send_daily_report():
    """Run daily at 9 AM."""
    report = await generate_report()
    await send_email("admin@example.com", report)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup - configure scheduler
    scheduler.add_job(
        cleanup_expired_sessions,
        CronTrigger(hour=0, minute=0),
        id="cleanup_sessions",
    )
    scheduler.add_job(
        send_daily_report,
        CronTrigger(hour=9, minute=0),
        id="daily_report",
    )
    scheduler.start()

    yield

    # Shutdown
    scheduler.shutdown()

app = FastAPI(lifespan=lifespan)


# Manual trigger endpoint (for testing)
@app.post("/admin/trigger/{job_id}")
async def trigger_job(job_id: str, admin: AdminUser):
    job = scheduler.get_job(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")

    job.modify(next_run_time=datetime.now())
    return {"message": f"Job {job_id} triggered"}
```

## Task Queues with Redis

```python
import redis.asyncio as redis
import json
from uuid import uuid4

class TaskQueue:
    def __init__(self, redis_url: str):
        self.redis = redis.from_url(redis_url)
        self.queue_name = "task_queue"

    async def enqueue(self, task_type: str, payload: dict) -> str:
        """Add task to queue."""
        task_id = str(uuid4())
        task = {
            "id": task_id,
            "type": task_type,
            "payload": payload,
            "status": "pending",
        }
        await self.redis.rpush(self.queue_name, json.dumps(task))
        await self.redis.set(f"task:{task_id}", json.dumps(task))
        return task_id

    async def get_status(self, task_id: str) -> dict | None:
        """Get task status."""
        data = await self.redis.get(f"task:{task_id}")
        return json.loads(data) if data else None


# Worker (separate process)
async def worker(queue: TaskQueue):
    """Process tasks from queue."""
    while True:
        task_data = await queue.redis.blpop(queue.queue_name, timeout=1)
        if not task_data:
            continue

        task = json.loads(task_data[1])
        task["status"] = "processing"
        await queue.redis.set(f"task:{task['id']}", json.dumps(task))

        try:
            result = await process_task(task)
            task["status"] = "completed"
            task["result"] = result
        except Exception as e:
            task["status"] = "failed"
            task["error"] = str(e)

        await queue.redis.set(f"task:{task['id']}", json.dumps(task))


# API endpoints
queue = TaskQueue("redis://localhost:6379")

@app.post("/tasks")
async def create_task(task_type: str, payload: dict):
    task_id = await queue.enqueue(task_type, payload)
    return {"task_id": task_id}

@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
    task = await queue.get_status(task_id)
    if not task:
        raise HTTPException(status_code=404)
    return task
```

## Async Task Manager

```python
import asyncio
from contextlib import asynccontextmanager
from typing import Callable, Awaitable

class TaskManager:
    """Manage long-running async tasks."""

    def __init__(self):
        self._tasks: dict[str, asyncio.Task] = {}
        self._results: dict[str, Any] = {}

    async def start(self, task_id: str, coro: Awaitable):
        """Start a named task."""
        if task_id in self._tasks:
            raise ValueError(f"Task {task_id} already running")

        async def wrapper():
            try:
                result = await coro
                self._results[task_id] = {"status": "completed", "result": result}
            except Exception as e:
                self._results[task_id] = {"status": "failed", "error": str(e)}
            finally:
                self._tasks.pop(task_id, None)

        self._tasks[task_id] = asyncio.create_task(wrapper())
        return task_id

    def get_status(self, task_id: str) -> dict:
        if task_id in self._tasks:
            return {"status": "running"}
        return self._results.get(task_id, {"status": "not_found"})

    async def cancel(self, task_id: str) -> bool:
        task = self._tasks.get(task_id)
        if task:
            task.cancel()
            return True
        return False


# Global task manager
task_manager = TaskManager()

@app.post("/process")
async def start_processing(data: ProcessRequest):
    task_id = str(uuid4())
    await task_manager.start(task_id, heavy_processing(data))
    return {"task_id": task_id}

@app.get("/process/{task_id}")
async def get_processing_status(task_id: str):
    return task_manager.get_status(task_id)
```

## Quick Reference

| Method | Use Case | Runs Where |
|--------|----------|------------|
| `BackgroundTasks` | Quick async tasks | Same process |
| Celery | Heavy processing | Worker process |
| APScheduler | Periodic jobs | Same process |
| Redis queue | Distributed tasks | Worker process |
| `asyncio.Task` | In-memory async | Same process |

| Pattern | Best For |
|---------|----------|
| BackgroundTasks | Email, webhooks, logging |
| Celery | Video processing, ML, reports |
| APScheduler | Cleanup, reports, sync |
| Redis queue | Scalable task distribution |

# FastAPI Dependency Injection Patterns

Advanced patterns for managing dependencies in FastAPI.

## Basic Dependencies

```python
from fastapi import Depends, FastAPI
from typing import Annotated

app = FastAPI()

# Simple dependency
async def get_db():
    db = DatabaseSession()
    try:
        yield db
    finally:
        await db.close()

# Use with Annotated for reusability
DB = Annotated[DatabaseSession, Depends(get_db)]

@app.get("/items")
async def get_items(db: DB):
    return await db.fetch_all("SELECT * FROM items")
```

## Dependency Hierarchy

```python
from fastapi import Depends, HTTPException, Header
from typing import Annotated

# Base dependency
async def get_db():
    async with async_session() as session:
        yield session

# Depends on get_db
async def get_current_user(
    db: Annotated[AsyncSession, Depends(get_db)],
    token: Annotated[str, Header()],
) -> User:
    user = await db.execute(
        select(User).where(User.token == token)
    )
    if not user:
        raise HTTPException(status_code=401)
    return user.scalar_one()

# Depends on get_current_user
async def get_admin_user(
    user: Annotated[User, Depends(get_current_user)],
) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=403)
    return user

# Reusable annotated types
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]
AdminUser = Annotated[User, Depends(get_admin_user)]

@app.get("/admin/users")
async def admin_list_users(admin: AdminUser, db: DB):
    return await db.execute(select(User)).scalars().all()
```

## Class-Based Dependencies

```python
from dataclasses import dataclass
from fastapi import Depends, Query
from typing import Annotated

@dataclass
class Pagination:
    """Reusable pagination parameters."""
    skip: int = 0
    limit: int = 10

    def __init__(
        self,
        skip: Annotated[int, Query(ge=0)] = 0,
        limit: Annotated[int, Query(ge=1, le=100)] = 10,
    ):
        self.skip = skip
        self.limit = limit

# Use as dependency
@app.get("/items")
async def list_items(pagination: Annotated[Pagination, Depends()]):
    return await fetch_items(
        skip=pagination.skip,
        limit=pagination.limit
    )


# Class with injected dependencies
class UserService:
    def __init__(self, db: Annotated[AsyncSession, Depends(get_db)]):
        self.db = db

    async def get_user(self, user_id: int) -> User | None:
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    service: Annotated[UserService, Depends()],
):
    user = await service.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404)
    return user
```

## Cached Dependencies

```python
from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    redis_url: str
    api_key: str

    model_config = {"env_file": ".env"}

@lru_cache
def get_settings() -> Settings:
    """Cached settings - loaded once."""
    return Settings()

# Use in dependencies
async def get_db(settings: Annotated[Settings, Depends(get_settings)]):
    engine = create_async_engine(settings.database_url)
    async with AsyncSession(engine) as session:
        yield session
```

## Request-Scoped State

```python
from fastapi import Request
from contextvars import ContextVar
from uuid import uuid4

# Context variable for request ID
request_id_var: ContextVar[str] = ContextVar("request_id")

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid4())
    request_id_var.set(request_id)
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

# Access in dependencies
def get_request_id() -> str:
    return request_id_var.get()

@app.get("/trace")
async def trace_request(request_id: Annotated[str, Depends(get_request_id)]):
    return {"request_id": request_id}
```

## Dependency Overrides for Testing

```python
from fastapi.testclient import TestClient

# Production dependency
async def get_db():
    async with async_session() as session:
        yield session

# Test override
async def get_test_db():
    async with test_session() as session:
        yield session

def test_create_user():
    app.dependency_overrides[get_db] = get_test_db

    with TestClient(app) as client:
        response = client.post("/users", json={"name": "Test"})
        assert response.status_code == 201

    app.dependency_overrides.clear()


# Context manager for cleaner tests
from contextlib import contextmanager

@contextmanager
def override_dependency(original, replacement):
    app.dependency_overrides[original] = replacement
    try:
        yield
    finally:
        app.dependency_overrides.pop(original, None)

def test_with_override():
    with override_dependency(get_db, get_test_db):
        # Test code here
        pass
```

## Parameterized Dependencies

```python
from fastapi import Depends
from typing import Callable

def require_permission(permission: str):
    """Factory for permission-checking dependencies."""
    async def check_permission(
        user: Annotated[User, Depends(get_current_user)],
    ):
        if permission not in user.permissions:
            raise HTTPException(
                status_code=403,
                detail=f"Missing permission: {permission}"
            )
        return user
    return check_permission

@app.delete("/items/{item_id}")
async def delete_item(
    item_id: int,
    user: Annotated[User, Depends(require_permission("items:delete"))],
):
    return {"deleted": item_id}


# Rate limiting factory
def rate_limit(requests: int, window: int):
    """Create rate limit dependency."""
    async def check_rate(
        request: Request,
        redis: Annotated[Redis, Depends(get_redis)],
    ):
        key = f"rate:{request.client.host}"
        count = await redis.incr(key)
        if count == 1:
            await redis.expire(key, window)
        if count > requests:
            raise HTTPException(status_code=429, detail="Rate limited")
    return check_rate

@app.get("/api/search")
async def search(
    q: str,
    _: Annotated[None, Depends(rate_limit(requests=100, window=60))],
):
    return {"query": q}
```

## Sub-Application Dependencies

```python
from fastapi import FastAPI, Depends

# Shared dependency for sub-app
async def get_api_key(x_api_key: Annotated[str, Header()]):
    if x_api_key != "secret":
        raise HTTPException(status_code=401)
    return x_api_key

# Sub-application with its own dependencies
api_v1 = FastAPI(dependencies=[Depends(get_api_key)])

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

# Mount on main app
app = FastAPI()
app.mount("/api/v1", api_v1)
```

## Quick Reference

| Pattern | Use Case |
|---------|----------|
| `Annotated[T, Depends(f)]` | Reusable dependency type |
| Class dependency | Group related params |
| `@lru_cache` | Cache settings/config |
| Dependency factory | Parameterized checks |
| `dependency_overrides` | Testing isolation |
| Hierarchy | Auth → User → Admin chain |
| `ContextVar` | Request-scoped state |

# FastAPI Middleware Patterns

Request/response processing, CORS, security, and error handling.

## Basic Middleware

```python
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# Function-based middleware
@app.middleware("http")
async def add_timing_header(request: 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}"
    return response


# Class-based middleware
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: 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}"
        return response

app.add_middleware(TimingMiddleware)
```

## CORS Configuration

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

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",
        "https://myapp.com",
    ],
    allow_credentials=True,
    allow_methods=["*"],  # Or specific: ["GET", "POST"]
    allow_headers=["*"],
    expose_headers=["X-Request-ID"],
    max_age=600,  # Cache preflight for 10 minutes
)

# Development: allow all origins
if settings.debug:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
```

## Security Headers

```python
from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)

        # Security headers
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

        # CSP - customize for your app
        response.headers["Content-Security-Policy"] = (
            "default-src 'self'; "
            "script-src 'self' 'unsafe-inline'; "
            "style-src 'self' 'unsafe-inline'"
        )

        # HSTS (only in production with HTTPS)
        if not request.url.scheme == "http":
            response.headers["Strict-Transport-Security"] = (
                "max-age=31536000; includeSubDomains"
            )

        return response

app.add_middleware(SecurityHeadersMiddleware)
```

## Request ID Tracking

```python
from uuid import uuid4
from contextvars import ContextVar

request_id_ctx: ContextVar[str] = ContextVar("request_id", default="")

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Use existing or generate new
        request_id = request.headers.get("X-Request-ID") or str(uuid4())

        # Store in context for logging
        request_id_ctx.set(request_id)
        request.state.request_id = request_id

        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id

        return response

app.add_middleware(RequestIDMiddleware)


# Access in endpoints
@app.get("/trace")
async def trace(request: Request):
    return {"request_id": request.state.request_id}
```

## Logging Middleware

```python
import logging
import time
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger(__name__)

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()

        # Log request
        logger.info(
            "Request started",
            extra={
                "method": request.method,
                "path": request.url.path,
                "client": request.client.host if request.client else None,
            }
        )

        response = await call_next(request)

        # Log response
        duration = time.perf_counter() - start
        logger.info(
            "Request completed",
            extra={
                "method": request.method,
                "path": request.url.path,
                "status": response.status_code,
                "duration": f"{duration:.3f}s",
            }
        )

        return response

app.add_middleware(LoggingMiddleware)
```

## Error Handling Middleware

```python
from fastapi import Request
from fastapi.responses import JSONResponse
import traceback

class ErrorHandlingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            return await call_next(request)
        except Exception as exc:
            # Log the full traceback
            logger.exception(
                "Unhandled exception",
                extra={
                    "path": request.url.path,
                    "method": request.method,
                    "traceback": traceback.format_exc(),
                }
            )

            # Return generic error (hide details in production)
            return JSONResponse(
                status_code=500,
                content={
                    "detail": "Internal server error",
                    "request_id": getattr(request.state, "request_id", None),
                },
            )

app.add_middleware(ErrorHandlingMiddleware)
```

## Rate Limiting

```python
from collections import defaultdict
from datetime import datetime, timedelta
import asyncio

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, requests: int = 100, window: int = 60):
        super().__init__(app)
        self.requests = requests
        self.window = window
        self.clients: dict[str, list[datetime]] = defaultdict(list)
        self.lock = asyncio.Lock()

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host if request.client else "unknown"
        now = datetime.now()
        window_start = now - timedelta(seconds=self.window)

        async with self.lock:
            # Remove old requests
            self.clients[client_ip] = [
                t for t in self.clients[client_ip]
                if t > window_start
            ]

            if len(self.clients[client_ip]) >= self.requests:
                return JSONResponse(
                    status_code=429,
                    content={"detail": "Rate limit exceeded"},
                    headers={
                        "Retry-After": str(self.window),
                        "X-RateLimit-Limit": str(self.requests),
                        "X-RateLimit-Remaining": "0",
                    },
                )

            self.clients[client_ip].append(now)
            remaining = self.requests - len(self.clients[client_ip])

        response = await call_next(request)
        response.headers["X-RateLimit-Limit"] = str(self.requests)
        response.headers["X-RateLimit-Remaining"] = str(remaining)
        return response

app.add_middleware(RateLimitMiddleware, requests=100, window=60)
```

## GZip Compression

```python
from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(
    GZipMiddleware,
    minimum_size=1000,  # Only compress responses > 1KB
)
```

## Trusted Host Validation

```python
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"],
)
```

## Middleware Order

```python
# Middleware executes in REVERSE order of addition
# Last added = First to process request, Last to process response

app = FastAPI()

# 1. Error handling (outermost - catches all errors)
app.add_middleware(ErrorHandlingMiddleware)

# 2. Logging (log after error handling)
app.add_middleware(LoggingMiddleware)

# 3. Request ID (needed for logging)
app.add_middleware(RequestIDMiddleware)

# 4. Security (before business logic)
app.add_middleware(SecurityHeadersMiddleware)

# 5. CORS (needs to be early for preflight)
app.add_middleware(CORSMiddleware, ...)

# 6. GZip (compress final response)
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Request flow: GZip → CORS → Security → RequestID → Logging → Error → App
# Response flow: App → Error → Logging → RequestID → Security → CORS → GZip
```

## Quick Reference

| Middleware | Purpose |
|------------|---------|
| `CORSMiddleware` | Cross-origin requests |
| `GZipMiddleware` | Response compression |
| `TrustedHostMiddleware` | Host validation |
| `BaseHTTPMiddleware` | Custom middleware base |
| `@app.middleware("http")` | Simple function middleware |

| Order Position | Middleware Type |
|----------------|-----------------|
| First (outer) | Error handling |
| Early | Logging, tracing |
| Middle | Auth, rate limiting |
| Late | CORS, compression |

# Pydantic v2 Validation & Serialization

Modern validation patterns for FastAPI with Pydantic v2.

## Basic Models

```python
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from typing import Annotated

class UserCreate(BaseModel):
    """Request model with field validation."""
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    bio: str | None = Field(default=None, max_length=500)

class UserResponse(BaseModel):
    """Response model with ORM support."""
    id: int
    name: str
    email: EmailStr
    created_at: datetime

    model_config = {"from_attributes": True}
```

## Custom Validators

```python
from pydantic import BaseModel, field_validator, model_validator
from typing import Self

class UserCreate(BaseModel):
    username: str
    password: str
    password_confirm: str

    @field_validator("username")
    @classmethod
    def validate_username(cls, v: str) -> str:
        """Validate single field."""
        if not v.isalnum():
            raise ValueError("Username must be alphanumeric")
        return v.lower()

    @model_validator(mode="after")
    def validate_passwords(self) -> Self:
        """Validate across multiple fields."""
        if self.password != self.password_confirm:
            raise ValueError("Passwords don't match")
        return self


# Before validation (raw input)
class Config(BaseModel):
    port: int

    @field_validator("port", mode="before")
    @classmethod
    def parse_port(cls, v):
        """Convert string to int before validation."""
        if isinstance(v, str):
            return int(v)
        return v
```

## Computed Fields

```python
from pydantic import BaseModel, computed_field
from datetime import datetime

class User(BaseModel):
    first_name: str
    last_name: str
    birth_date: datetime

    @computed_field
    @property
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

    @computed_field
    @property
    def age(self) -> int:
        today = datetime.now()
        return today.year - self.birth_date.year
```

## Field Serialization

```python
from pydantic import BaseModel, field_serializer
from datetime import datetime
from decimal import Decimal

class Order(BaseModel):
    id: int
    total: Decimal
    created_at: datetime

    @field_serializer("total")
    def serialize_total(self, value: Decimal) -> str:
        """Serialize Decimal as formatted string."""
        return f"${value:.2f}"

    @field_serializer("created_at")
    def serialize_date(self, value: datetime) -> str:
        """Serialize datetime as ISO string."""
        return value.isoformat()


# Or use Annotated with serialization
from pydantic import PlainSerializer

FormattedDecimal = Annotated[
    Decimal,
    PlainSerializer(lambda v: f"${v:.2f}", return_type=str)
]

class Order(BaseModel):
    total: FormattedDecimal
```

## Custom Types

```python
from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import CoreSchema, core_schema
from typing import Any

class PhoneNumber(str):
    """Custom phone number type with validation."""

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: GetCoreSchemaHandler
    ) -> CoreSchema:
        return core_schema.no_info_after_validator_function(
            cls._validate,
            core_schema.str_schema(),
        )

    @classmethod
    def _validate(cls, v: str) -> "PhoneNumber":
        # Remove non-digits
        digits = "".join(c for c in v if c.isdigit())
        if len(digits) != 10:
            raise ValueError("Phone must be 10 digits")
        return cls(f"({digits[:3]}) {digits[3:6]}-{digits[6:]}")


class Contact(BaseModel):
    name: str
    phone: PhoneNumber

# Usage
contact = Contact(name="John", phone="1234567890")
print(contact.phone)  # (123) 456-7890
```

## Nested Models

```python
from pydantic import BaseModel
from datetime import datetime

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

class Company(BaseModel):
    name: str
    address: Address

class UserResponse(BaseModel):
    id: int
    name: str
    company: Company | None = None
    addresses: list[Address] = []

    model_config = {"from_attributes": True}
```

## Discriminated Unions

```python
from pydantic import BaseModel, Field
from typing import Literal, Union
from typing_extensions import Annotated

class Dog(BaseModel):
    pet_type: Literal["dog"]
    name: str
    breed: str

class Cat(BaseModel):
    pet_type: Literal["cat"]
    name: str
    indoor: bool = True

# Use discriminator for efficient parsing
Pet = Annotated[
    Union[Dog, Cat],
    Field(discriminator="pet_type")
]

class Owner(BaseModel):
    name: str
    pets: list[Pet]

# FastAPI automatically validates
@app.post("/owners")
async def create_owner(owner: Owner):
    return owner
```

## Model Inheritance

```python
from pydantic import BaseModel
from datetime import datetime

class BaseResponse(BaseModel):
    """Base for all responses."""
    model_config = {"from_attributes": True}

class TimestampMixin(BaseModel):
    """Mixin for timestamp fields."""
    created_at: datetime
    updated_at: datetime

class UserBase(BaseModel):
    name: str
    email: str

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase, TimestampMixin, BaseResponse):
    id: int
```

## Partial Updates (PATCH)

```python
from pydantic import BaseModel
from typing import Any

class UserUpdate(BaseModel):
    """All fields optional for partial updates."""
    name: str | None = None
    email: str | None = None
    bio: str | None = None

@app.patch("/users/{user_id}")
async def update_user(user_id: int, updates: UserUpdate):
    # Only get set fields
    update_data = updates.model_dump(exclude_unset=True)

    # Apply to existing user
    user = await get_user(user_id)
    for field, value in update_data.items():
        setattr(user, field, value)

    return user
```

## Validation Error Handling

```python
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request,
    exc: RequestValidationError
):
    """Custom validation error response."""
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
        })

    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "errors": errors,
        },
    )
```

## Quick Reference

| Feature | Pydantic v2 |
|---------|-------------|
| ORM mode | `model_config = {"from_attributes": True}` |
| Field validator | `@field_validator("field")` |
| Model validator | `@model_validator(mode="after")` |
| Serializer | `@field_serializer("field")` |
| Computed | `@computed_field` + `@property` |
| Exclude unset | `model_dump(exclude_unset=True)` |
| Discriminator | `Field(discriminator="type")` |

| Validation | Usage |
|------------|-------|
| Required | `name: str` |
| Optional | `name: str \| None = None` |
| Default | `name: str = "default"` |
| Constraints | `Field(min_length=1, max_length=100)` |
| Custom | `@field_validator` |

#!/usr/bin/env bash
# Generate FastAPI endpoint boilerplate (Pydantic models + CRUD router) to stdout.
#
# Usage:   scaffold-api.sh <resource_name>
# Input:   one positional arg — a singular resource name (e.g. "user", "order")
# Output:  a complete FastAPI module (models + APIRouter CRUD endpoints) on stdout;
#          redirect into a file to use it (the script emits to stdout and never
#          opens a file itself, so it cannot clobber a user's project)
# Stderr:  usage and error messages only
# Exit:    0 ok, 2 usage (missing resource / unknown flag)
#
# Examples:
#   scaffold-api.sh user > routers/user.py
#   scaffold-api.sh order | tee routers/orders.py

set -uo pipefail

usage() {
  cat <<'EOF'
Usage: scaffold-api.sh <resource_name>

Generate FastAPI endpoint boilerplate — Pydantic Create/Update/Response models
plus an APIRouter with list/create/get/update/delete CRUD endpoints — printed to
stdout. Redirect into a module file to use it.

Arguments:
  resource_name   a singular resource name, e.g. "user" or "order"

Exit codes: 0 ok, 2 usage (missing resource / unknown flag).

Examples:
  scaffold-api.sh user > routers/user.py
  scaffold-api.sh order | tee routers/orders.py
EOF
}

# Flags first: --help short-circuits; any unknown flag is a hard usage error.
while [[ $# -gt 0 ]]; do
  case "$1" in
    -h|--help) usage; exit 0 ;;
    --) shift; break ;;
    -*) echo "scaffold-api.sh: unknown option: $1" >&2; usage >&2; exit 2 ;;
    *) break ;;
  esac
done

RESOURCE="${1:-}"

if [[ -z "$RESOURCE" ]]; then
  echo "scaffold-api.sh: resource name required" >&2
  usage >&2
  exit 2
fi

# Convert to different cases
RESOURCE_LOWER=$(echo "$RESOURCE" | tr '[:upper:]' '[:lower:]')
RESOURCE_UPPER=$(echo "$RESOURCE" | tr '[:lower:]' '[:upper:]')
RESOURCE_TITLE=$(echo "$RESOURCE_LOWER" | sed 's/\b\(.\)/\u\1/g')
RESOURCE_PLURAL="${RESOURCE_LOWER}s"

cat << EOF
# =============================================================================
# ${RESOURCE_TITLE} Models
# =============================================================================

from pydantic import BaseModel, Field
from datetime import datetime

class ${RESOURCE_TITLE}Create(BaseModel):
    """Create ${RESOURCE_LOWER} request."""
    name: str = Field(..., min_length=1, max_length=100)
    # Add more fields

class ${RESOURCE_TITLE}Update(BaseModel):
    """Update ${RESOURCE_LOWER} request (partial)."""
    name: str | None = None
    # Add more fields

class ${RESOURCE_TITLE}Response(BaseModel):
    """${RESOURCE_TITLE} response."""
    id: int
    name: str
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}


# =============================================================================
# ${RESOURCE_TITLE} Router
# =============================================================================

from fastapi import APIRouter, Depends, HTTPException
from typing import Annotated

router = APIRouter(prefix="/${RESOURCE_PLURAL}", tags=["${RESOURCE_PLURAL}"])

@router.get("/", response_model=list[${RESOURCE_TITLE}Response])
async def list_${RESOURCE_PLURAL}(
    db: DB,
    skip: int = 0,
    limit: int = 10,
):
    """List all ${RESOURCE_PLURAL}."""
    result = await db.execute(
        select(${RESOURCE_TITLE}).offset(skip).limit(limit)
    )
    return result.scalars().all()

@router.post("/", response_model=${RESOURCE_TITLE}Response, status_code=201)
async def create_${RESOURCE_LOWER}(data: ${RESOURCE_TITLE}Create, db: DB):
    """Create a new ${RESOURCE_LOWER}."""
    ${RESOURCE_LOWER} = ${RESOURCE_TITLE}(**data.model_dump())
    db.add(${RESOURCE_LOWER})
    await db.commit()
    await db.refresh(${RESOURCE_LOWER})
    return ${RESOURCE_LOWER}

@router.get("/{${RESOURCE_LOWER}_id}", response_model=${RESOURCE_TITLE}Response)
async def get_${RESOURCE_LOWER}(${RESOURCE_LOWER}_id: int, db: DB):
    """Get a ${RESOURCE_LOWER} by ID."""
    ${RESOURCE_LOWER} = await db.get(${RESOURCE_TITLE}, ${RESOURCE_LOWER}_id)
    if not ${RESOURCE_LOWER}:
        raise HTTPException(status_code=404, detail="${RESOURCE_TITLE} not found")
    return ${RESOURCE_LOWER}

@router.patch("/{${RESOURCE_LOWER}_id}", response_model=${RESOURCE_TITLE}Response)
async def update_${RESOURCE_LOWER}(
    ${RESOURCE_LOWER}_id: int,
    data: ${RESOURCE_TITLE}Update,
    db: DB,
):
    """Update a ${RESOURCE_LOWER}."""
    ${RESOURCE_LOWER} = await db.get(${RESOURCE_TITLE}, ${RESOURCE_LOWER}_id)
    if not ${RESOURCE_LOWER}:
        raise HTTPException(status_code=404, detail="${RESOURCE_TITLE} not found")

    for field, value in data.model_dump(exclude_unset=True).items():
        setattr(${RESOURCE_LOWER}, field, value)

    await db.commit()
    await db.refresh(${RESOURCE_LOWER})
    return ${RESOURCE_LOWER}

@router.delete("/{${RESOURCE_LOWER}_id}", status_code=204)
async def delete_${RESOURCE_LOWER}(${RESOURCE_LOWER}_id: int, db: DB):
    """Delete a ${RESOURCE_LOWER}."""
    ${RESOURCE_LOWER} = await db.get(${RESOURCE_TITLE}, ${RESOURCE_LOWER}_id)
    if not ${RESOURCE_LOWER}:
        raise HTTPException(status_code=404, detail="${RESOURCE_TITLE} not found")

    await db.delete(${RESOURCE_LOWER})
    await db.commit()

# =============================================================================
# Include in main app:
# from routers.${RESOURCE_PLURAL} import router as ${RESOURCE_PLURAL}_router
# app.include_router(${RESOURCE_PLURAL}_router, prefix="/api/v1")
# =============================================================================
EOF

#!/usr/bin/env bash
# Self-test for python-fastapi-ops — fully offline, deterministic, Linux-safe.
#
# scaffold-api.sh emits a FastAPI module to stdout; it never opens a file for
# writing, so it cannot clobber a user's project. Every run here is redirected
# into a mktemp -d sandbox — nothing is ever written into the repo or a real
# project. Asserts the protocol contract (--help, exit codes, stream
# separation), the generated module's structure, byte-identical idempotency,
# and that the script owns no overwrite hazard (the only clobber is the
# caller's shell `>` redirect — a latent caller-side hazard, documented here,
# NOT "fixed" in the script per its contract).
#
# Usage:   bash tests/run.sh
# Exit:    0 all pass, 1 one or more failures

set -uo pipefail

HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKILL="$(dirname "$HERE")"
V="$SKILL/scripts/scaffold-api.sh"

SB="$(mktemp -d)"; trap 'rm -rf "$SB"' EXIT
PASS=0; FAIL=0
ok() { PASS=$((PASS+1)); printf '  PASS  %s\n' "$1"; }
no() { FAIL=$((FAIL+1)); printf '  FAIL  %s\n' "$1"; }
expect_exit() { [[ "$2" == "$3" ]] && ok "$1 (exit $3)" || no "$1 (want $2 got $3)"; }
expect_has()  { case "$3" in *"$2"*) ok "$1";; *) no "$1 (missing '$2')";; esac; }

echo "=== python-fastapi-ops self-test ==="

# ── contract ──────────────────────────────────────────────────────────────────
echo "-- contract --"
bash -n "$V" 2>/dev/null && ok "bash -n scaffold-api.sh" || no "bash -n scaffold-api.sh"
bash "$V" --help >/dev/null 2>&1; expect_exit "--help exits 0" 0 $?
bash "$V" -h     >/dev/null 2>&1; expect_exit "-h exits 0" 0 $?
out="$(bash "$V" --help 2>/dev/null)"
expect_has "--help has Examples" "xamples" "$out"
expect_has "--help documents exit 2" "2" "$out"
bash "$V"          >/dev/null 2>&1; expect_exit "missing resource -> 2" 2 $?
bash "$V" --bogus  >/dev/null 2>&1; expect_exit "unknown flag -> 2" 2 $?

# usage/errors must go to stderr, never polluting the stdout data stream
err="$(bash "$V" 2>&1 >/dev/null)"
expect_has "missing-resource usage on stderr" "Usage:" "$err"
noso="$(bash "$V" --bogus 2>/dev/null)"
[[ -z "$noso" ]] && ok "unknown-flag writes nothing to stdout" || no "unknown-flag leaked to stdout"

# ── generated module structure ────────────────────────────────────────────────
echo "-- generated module --"
bash "$V" user >"$SB/user.py" 2>"$SB/user.err"; expect_exit "generate user -> 0" 0 $?
[[ -s "$SB/user.py" ]] && ok "module written to redirected stdout" || no "module not written"
[[ ! -s "$SB/user.err" ]] && ok "happy path is silent on stderr" || no "happy path wrote to stderr"
out="$(cat "$SB/user.py")"
expect_has "has Pydantic Create model"  "class UserCreate(BaseModel):" "$out"
expect_has "has Pydantic Update model"  "class UserUpdate(BaseModel):" "$out"
expect_has "has Pydantic Response model" "class UserResponse(BaseModel):" "$out"
expect_has "has APIRouter with prefix"  'router = APIRouter(prefix="/users"' "$out"
expect_has "has create endpoint"  "async def create_user" "$out"
expect_has "has list endpoint"    "async def list_users" "$out"
expect_has "has get endpoint"     "async def get_user" "$out"
expect_has "has update endpoint"  "async def update_user" "$out"
expect_has "has delete endpoint"  "async def delete_user" "$out"
expect_has "resource name in Create docstring" "Create user request" "$out"

# title-casing + pluralization for mixed-case input
bash "$V" Order >"$SB/order.py" 2>/dev/null; expect_exit "generate Order -> 0" 0 $?
out="$(cat "$SB/order.py")"
expect_has "title-cases Order"        "class OrderCreate(BaseModel):" "$out"
expect_has "pluralizes to orders"     'prefix="/orders"' "$out"
expect_has "orders create endpoint"   "async def create_order" "$out"

# ── idempotency: re-running produces byte-identical output ───────────────────
echo "-- idempotency --"
bash "$V" user >"$SB/a.py" 2>/dev/null
bash "$V" user >"$SB/b.py" 2>/dev/null
if cmp -s "$SB/a.py" "$SB/b.py"; then ok "re-run produces byte-identical output"; else no "re-run output differs"; fi

# ── overwrite safety ─────────────────────────────────────────────────────────
echo "-- overwrite safety --"
# The script emits to stdout only and never opens a file, so it is
# non-destructive by construction: running it leaves the cwd untouched. There
# is no --force/--no-clobber flag because the script has no file target to
# guard — the sole clobber path is the caller's own shell redirect (`>`),
# which the script cannot see. That caller-side hazard is documented here,
# not "fixed".
mkdir -p "$SB/cwd-check"
( cd "$SB/cwd-check" && bash "$V" user >/dev/null 2>&1 )
n="$(find "$SB/cwd-check" -type f | wc -l | tr -d ' ')"
[[ "$n" == "0" ]] && ok "script creates no files in cwd (non-destructive)" || no "script wrote files into cwd ($n)"

echo ""
echo "=== $PASS passed, $FAIL failed ==="
[[ "$FAIL" -eq 0 ]] || exit 1
exit 0