FastAPI: The Complete Guide 2026 Edition

# Type parameter syntax (PEP 695)
type UserID = int
type JSONResponse[T] = dict[str, T]
 
# Improved error messages
# Error messages now point exactly to the issue
 
# F-strings in more contexts
print(f"{user.name = }")  # Debugging made easier

from pydantic import BaseModel, ConfigDict, field_validator
from typing import Annotated
from annotated_types import Gt, Lt
 
class ProductCreate(BaseModel):
    model_config = ConfigDict(
        strict=True,  # No type coercion
        frozen=True,  # Immutable instances
        extra='forbid'  # Reject unknown fields
    )
    
    name: str
    price: Annotated[float, Gt(0), Lt(10000)]
    quantity: int = 1
    
    @field_validator('name')
    @classmethod
    def validate_name(cls, v: str) -> str:
        return v.strip().title()

 
src/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application factory
│   ├── config.py            # Settings management
│   │
│   ├── core/                # Shared infrastructure
│   │   ├── security.py
│   │   ├── database.py
│   │   └── exceptions.py
│   │
│   ├── features/            # Domain-driven design
│   │   ├── users/
│   │   │   ├── router.py
│   │   │   ├── service.py
│   │   │   ├── repository.py
│   │   │   ├── models.py
│   │   │   └── schemas.py
│   │   │
│   │   └── products/
│   │       └── ...
│   │
│   └── shared/              # Cross-cutting concerns
│       ├── middleware.py
│       └── utils.py
│
├── tests/
├── alembic/
├── docker/
└── pyproject.toml

from typing import Annotated
from fastapi import Depends, Query, Path
from sqlalchemy.ext.asyncio import AsyncSession
 
# Define reusable dependency types
DBSession = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_active_user)]
AdminUser = Annotated[User, Depends(require_admin)]
 
# Pagination with validation
def get_pagination(
    page: Annotated[int, Query(ge=1, description="Page number")] = 1,
    size: Annotated[int, Query(ge=1, le=100, description="Items per page")] = 20
) -> dict:
    return {"skip": (page - 1) * size, "limit": size}
 
Pagination = Annotated[dict, Depends(get_pagination)]
 
# Clean endpoint signatures
@router.get("/users")
async def list_users(
    db: DBSession,
    pagination: Pagination,
    current_user: AdminUser
) -> PagedResponse[UserResponse]:
    pass

import asyncio
from typing import List
 
async def fetch_with_timeout(url: str, timeout: float = 5.0):
    """Fetch URL with timeout protection."""
    async with asyncio.timeout(timeout):
        async with httpx.AsyncClient() as client:
            return await client.get(url)
 
async def fetch_all_or_nothing(urls: List[str]):
    """Fetch all URLs or fail completely using TaskGroup."""
    results = []
    
    async with asyncio.TaskGroup() as tg:
        tasks = [
            tg.create_task(fetch_with_timeout(url))
            for url in urls
        ]
    
    # All tasks succeeded if we reach here
    return [task.result() for task in tasks]
 
@app.get("/aggregate")
async def aggregate_data():
    """Aggregate data from multiple sources."""
    try:
        responses = await fetch_all_or_nothing([
            "https://api1.example.com/data",
            "https://api2.example.com/data",
            "https://api3.example.com/data"
        ])
        return {"data": [r.json() for r in responses]}
    except* httpx.TimeoutException:
        raise HTTPException(503, "External service timeout")

import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List
 
@dataclass
class SlidingWindowCounter:
    window_size: int = 60  # seconds
    max_requests: int = 100
    requests: Dict[str, List[float]] = field(default_factory=lambda: defaultdict(list))
    
    def is_allowed(self, key: str) -> bool:
        now = time.time()
        window_start = now - self.window_size
        
        # Clean old requests
        self.requests[key] = [
            ts for ts in self.requests[key] 
            if ts > window_start
        ]
        
        if len(self.requests[key]) >= self.max_requests:
            return False
        
        self.requests[key].append(now)
        return True
    
    def get_remaining(self, key: str) -> int:
        now = time.time()
        window_start = now - self.window_size
        current = len([ts for ts in self.requests[key] if ts > window_start])
        return max(0, self.max_requests - current)
 
rate_limiter = SlidingWindowCounter(max_requests=100)
 
async def rate_limit_dependency(request: Request):
    client_ip = request.client.host
    
    if not rate_limiter.is_allowed(client_ip):
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded",
            headers={
                "Retry-After": "60",
                "X-RateLimit-Remaining": "0"
            }
        )
    
    return rate_limiter.get_remaining(client_ip)

from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 
# Setup tracing
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
 
# Instrument FastAPI
FastAPIInstrumentor.instrument_app(app)
 
tracer = trace.get_tracer(__name__)
 
@app.get("/orders/{order_id}")
async def get_order(order_id: int, db: DBSession):
    with tracer.start_as_current_span("fetch_order") as span:
        span.set_attribute("order.id", order_id)
        
        order = await db.get(Order, order_id)
        
        span.set_attribute("order.status", order.status)
        span.set_attribute("order.items_count", len(order.items))
        
        return order

# syntax=docker/dockerfile:1.4
FROM python:3.12-slim as base
 
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1 \
    PIP_DISABLE_PIP_VERSION_CHECK=1
 
FROM base as builder
 
RUN pip install uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev
 
FROM base as production
 
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY src/ /app/src/
 
WORKDIR /app
 
# Non-root user for security
RUN adduser --disabled-password --gecos "" appuser
USER appuser
 
EXPOSE 8000
CMD ["python", "-m", "uvicorn", "src.app.main:app", "--host", "0.0.0.0"]

# Optimized for Cloudflare Workers / Vercel Edge
from fastapi import FastAPI
from mangum import Mangum  # AWS Lambda adapter
 
app = FastAPI()
 
# Edge-optimized caching
@app.get("/api/config")
async def get_config():
    return Response(
        content=json.dumps(CONFIG),
        headers={
            "Cache-Control": "public, s-maxage=3600, stale-while-revalidate=86400",
            "CDN-Cache-Control": "max-age=7200"
        }
    )
 
# Lambda handler
handler = Mangum(app, lifespan="off")

from pydantic import BaseModel
 
class AIGeneratedEndpoint(BaseModel):
    """Schema for AI-generated API specifications."""
    path: str
    method: str
    summary: str
    request_schema: dict
    response_schema: dict
 
def register_dynamic_endpoint(spec: AIGeneratedEndpoint):
    """Dynamically register endpoints from AI specifications."""
    
    async def handler(request: Request):
        # Implementation generated or configured
        pass
    
    app.add_api_route(
        spec.path,
        handler,
        methods=[spec.method],
        summary=spec.summary
    )

from fastapi import Security
from fastapi.security import APIKeyHeader
from cryptography.fernet import Fernet
import secrets
 
# API key validation with constant-time comparison
api_key_header = APIKeyHeader(name="X-API-Key")
 
async def verify_api_key(
    api_key: str = Security(api_key_header)
):
    expected = get_api_key_from_db()
    if not secrets.compare_digest(api_key, expected):
        raise HTTPException(401, "Invalid API key")
 
# Field-level encryption for sensitive data
class EncryptedField:
    def __init__(self, key: bytes):
        self.fernet = Fernet(key)
    
    def encrypt(self, value: str) -> str:
        return self.fernet.encrypt(value.encode()).decode()
    
    def decrypt(self, value: str) -> str:
        return self.fernet.decrypt(value.encode()).decode()

# Connection pooling with SQLAlchemy 2.0
from sqlalchemy.ext.asyncio import create_async_engine
 
engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    pool_recycle=3600
)
 
# Response compression
from fastapi.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)
 
# Efficient serialization
from orjson import dumps
 
@app.get("/large-dataset")
async def get_large_dataset():
    data = await fetch_large_data()
    return Response(
        content=dumps(data),
        media_type="application/json"
    )