# FastAPI Best Practices

> **A comprehensive guide for building production-grade FastAPI applications.**
> Based on patterns from [zhanymkanov/fastapi-best-practices](https://github.com/zhanymkanov/fastapi-best-practices) with additional patterns for logging, authentication, and scalable architecture.

---

## 1. Project Structure

### Domain-Based Organization (NOT File-Type Based)

**Do NOT** group files by type (e.g., one huge `routers/` folder). Group by **domain/module** inside a `src/` directory.

```
project/
├── AGENTS.md                    # AI/developer guidelines
├── README.md                    # Project overview
├── CHANGELOG.md                 # Version history
├── pyproject.toml               # Package metadata
├── requirements.txt             # Production dependencies only
├── requirements-dev.txt         # Dev/test dependencies
├── .env.example                 # Environment template
│
├── src/
│   ├── __init__.py
│   ├── main.py                  # FastAPI app, lifespan (NO routes here)
│   │
│   ├── shared/                  # Cross-cutting concerns
│   │   ├── __init__.py
│   │   ├── base.py              # BaseController, BaseSchema
│   │   ├── config.py            # Pydantic BaseSettings
│   │   ├── logging.py           # Logger decorator + centralized setup
│   │   ├── exceptions.py        # Custom exception hierarchy
│   │   ├── auth.py              # Authentication utilities
│   │   └── context.py           # Request context (user provider, etc.)
│   │
│   └── domains/                 # Feature domains
│       ├── __init__.py
│       ├── router.py            # Root router - composes all domain routers
│       │
│       ├── health/              # Health check domain
│       │   ├── __init__.py
│       │   ├── router.py        # Routes
│       │   └── controller.py    # Business logic
│       │
│       └── {domain}/            # Each feature domain
│           ├── __init__.py
│           ├── router.py        # Domain routes
│           ├── controller.py    # Business logic
│           ├── schemas.py       # Pydantic models
│           ├── service.py       # External service calls (optional)
│           └── exceptions.py    # Domain-specific exceptions (optional)
│
├── tests/
│   ├── __init__.py
│   ├── conftest.py              # Pytest fixtures
│   └── test_{domain}.py
│
└── docs/
    └── architecture.md
```

### Key Principles

1. **Clean main.py**: Only app creation, middleware, lifespan. NO routes.
2. **Domain routers**: Each domain has `router.py`. Root `domains/router.py` composes them.
3. **Shared utilities**: Cross-cutting concerns in `shared/` - import from there, not across domains.
4. **Self-contained domains**: Each domain can be understood in isolation.

---

## 2. Base Patterns

### BaseController

```python
from abc import ABC, abstractmethod
from fastapi import APIRouter


class BaseController(ABC):
    """
    Base controller with lazy router instantiation.

    All domain controllers inherit from this and implement create_router().
    """

    def __init__(self, prefix: str, tags: list[str]):
        self.prefix = prefix
        self.tags = tags
        self._router = None

    @abstractmethod
    def create_router(self) -> APIRouter:
        """Create and configure the FastAPI router with all routes."""
        pass

    @property
    def router(self) -> APIRouter:
        """Lazy router instantiation."""
        if self._router is None:
            self._router = self.create_router()
        return self._router
```

**Usage:**
```python
# src/domains/health/controller.py
class HealthController(BaseController):
    def __init__(self):
        super().__init__(prefix="", tags=["Health"])

    def create_router(self) -> APIRouter:
        router = APIRouter(tags=self.tags)

        @router.get("/health")
        async def health_check():
            return {"status": "healthy"}

        return router

health_controller = HealthController()
```

### BaseSchema

```python
from datetime import datetime
from typing import Any
from pydantic import BaseModel, ConfigDict


class BaseSchema(BaseModel):
    """Base Pydantic model with standardized configuration."""

    model_config = ConfigDict(
        strict=False,
        populate_by_name=True,
        use_enum_values=True,
        validate_assignment=True,
        json_encoders={
            datetime: lambda v: v.isoformat() if v else None
        }
    )

    def dict_without_none(self) -> dict[str, Any]:
        """Return model as dict, excluding None values."""
        return {k: v for k, v in self.model_dump().items() if v is not None}
```

---

## 3. Configuration

### Pydantic Settings

```python
# src/shared/config.py
import tomllib
from pathlib import Path
from functools import lru_cache
from pydantic_settings import BaseSettings


def _get_version() -> str:
    """Load version from pyproject.toml."""
    try:
        with open(Path(__file__).parent.parent.parent / "pyproject.toml", "rb") as f:
            return tomllib.load(f).get("project", {}).get("version", "0.0.0")
    except FileNotFoundError:
        return "0.0.0"


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

    # Application
    app_name: str = "MyApp"
    app_version: str = _get_version()
    debug: bool = False

    # Server
    host: str = "0.0.0.0"
    port: int = 8000

    # Logging
    log_level: str = "INFO"

    # CORS
    cors_origins: list[str] = ["http://localhost:3000"]
    cors_credentials: bool = True
    cors_methods: list[str] = ["*"]
    cors_headers: list[str] = ["*"]

    class Config:
        env_file = ".env"
        case_sensitive = False
        extra = "ignore"


@lru_cache()
def get_settings() -> Settings:
    """Cached settings singleton."""
    return Settings()
```

---

## 4. Logging with Temporal Benchmarking

### Logger Decorator

```python
# src/shared/logging.py
import functools
import asyncio
import time
import logging
import sys
from pathlib import Path
from typing import Callable, Optional
from contextvars import ContextVar
from dataclasses import dataclass, field
from uuid import uuid4


# === Trace Context ===

@dataclass
class TraceSpan:
    """Represents a timed execution span."""
    name: str
    trace_id: str
    parent_id: Optional[str] = None
    span_id: str = field(default_factory=lambda: uuid4().hex[:8])
    start_time: float = field(default_factory=time.perf_counter)
    end_time: Optional[float] = None

    @property
    def duration_ms(self) -> float:
        end = self.end_time or time.perf_counter()
        return (end - self.start_time) * 1000


_current_span: ContextVar[Optional[TraceSpan]] = ContextVar('current_span', default=None)
_trace_id: ContextVar[Optional[str]] = ContextVar('trace_id', default=None)


def get_current_trace_id() -> Optional[str]:
    """Get current trace ID for log correlation."""
    return _trace_id.get()


# === Setup ===

def setup_logging(log_level: str = "INFO") -> None:
    """Configure application logging."""
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)

    logging.basicConfig(
        level=getattr(logging, log_level.upper()),
        format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        handlers=[
            logging.StreamHandler(sys.stdout),
            logging.FileHandler(log_dir / "app.log", encoding="utf-8")
        ]
    )

    # Quiet noisy libraries
    for name in ["httpx", "httpcore", "uvicorn.access"]:
        logging.getLogger(name).setLevel(logging.WARNING)


def get_logger(name: str) -> logging.Logger:
    """Get a logger instance."""
    return logging.getLogger(name)


# === Decorator ===

def logged(
    logger: logging.Logger = None,
    slow_threshold_ms: float = 100.0,
    warn_threshold_ms: float = 500.0,
    include_args: bool = False,
):
    """
    Decorator for automatic function logging with temporal benchmarking.

    Args:
        logger: Logger instance (defaults to module logger)
        slow_threshold_ms: Log INFO if execution exceeds this (default 100ms)
        warn_threshold_ms: Log WARNING if execution exceeds this (default 500ms)
        include_args: Include function arguments in log (careful with sensitive data)

    Usage:
        @logged()
        async def my_function(): ...

        @logged(slow_threshold_ms=50, warn_threshold_ms=200)
        def critical_path(): ...
    """
    def decorator(func: Callable):
        nonlocal logger
        if logger is None:
            logger = logging.getLogger(func.__module__)

        func_name = f"{func.__module__}.{func.__qualname__}"

        def _create_span() -> TraceSpan:
            parent = _current_span.get()
            trace_id = _trace_id.get() or uuid4().hex[:16]
            if _trace_id.get() is None:
                _trace_id.set(trace_id)
            return TraceSpan(
                name=func_name,
                trace_id=trace_id,
                parent_id=parent.span_id if parent else None,
            )

        def _log_completion(span: TraceSpan, error: Exception = None):
            span.end_time = time.perf_counter()
            duration = span.duration_ms
            tid = span.trace_id[:8]

            if error:
                logger.error(f"[{tid}] {func_name} FAILED after {duration:.2f}ms: {error}", exc_info=True)
            elif duration >= warn_threshold_ms:
                logger.warning(f"[{tid}] {func_name} SLOW: {duration:.2f}ms (threshold: {warn_threshold_ms}ms)")
            elif duration >= slow_threshold_ms:
                logger.info(f"[{tid}] {func_name} completed in {duration:.2f}ms")
            else:
                logger.debug(f"[{tid}] {func_name} completed in {duration:.2f}ms")

        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs):
            span = _create_span()
            token = _current_span.set(span)
            logger.debug(f"[{span.trace_id[:8]}] -> {func_name}")
            try:
                result = await func(*args, **kwargs)
                _log_completion(span)
                return result
            except Exception as e:
                _log_completion(span, error=e)
                raise
            finally:
                _current_span.reset(token)

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            span = _create_span()
            token = _current_span.set(span)
            logger.debug(f"[{span.trace_id[:8]}] -> {func_name}")
            try:
                result = func(*args, **kwargs)
                _log_completion(span)
                return result
            except Exception as e:
                _log_completion(span, error=e)
                raise
            finally:
                _current_span.reset(token)

        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    return decorator


# === Context Manager ===

class trace_span:
    """
    Context manager for manual span creation.

    Usage:
        with trace_span("database_query"):
            result = db.execute(query)

        async with trace_span("external_api_call"):
            response = await client.get(url)
    """
    def __init__(self, name: str, logger: logging.Logger = None):
        self.name = name
        self.logger = logger or logging.getLogger(__name__)
        self.span: Optional[TraceSpan] = None
        self.token = None

    def __enter__(self):
        parent = _current_span.get()
        trace_id = _trace_id.get() or uuid4().hex[:16]
        if _trace_id.get() is None:
            _trace_id.set(trace_id)

        self.span = TraceSpan(
            name=self.name,
            trace_id=trace_id,
            parent_id=parent.span_id if parent else None,
        )
        self.token = _current_span.set(self.span)
        self.logger.debug(f"[{self.span.trace_id[:8]}] -> {self.name}")
        return self.span

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.span:
            self.span.end_time = time.perf_counter()
            duration = self.span.duration_ms
            tid = self.span.trace_id[:8]
            if exc_val:
                self.logger.error(f"[{tid}] {self.name} FAILED: {duration:.2f}ms")
            else:
                self.logger.debug(f"[{tid}] {self.name}: {duration:.2f}ms")
        if self.token:
            _current_span.reset(self.token)
        return False

    async def __aenter__(self):
        return self.__enter__()

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        return self.__exit__(exc_type, exc_val, exc_tb)
```

**Example output:**
```
DEBUG [a1b2c3d4] -> src.domains.users.controller.get_user
DEBUG [a1b2c3d4] -> src.domains.users.service.fetch_from_db
DEBUG [a1b2c3d4] src.domains.users.service.fetch_from_db: 12.34ms
DEBUG [a1b2c3d4] src.domains.users.controller.get_user completed in 15.67ms
WARN  [a1b2c3d4] src.domains.reports.controller.generate SLOW: 523.45ms
```

---

## 5. Request Context (User Provider)

### Singleton Pattern with ContextVars

```python
# src/shared/context.py
from dataclasses import dataclass
from typing import Optional
from contextvars import ContextVar


@dataclass
class User:
    """Authenticated user context."""
    id: str
    email: str
    tenant_id: Optional[str] = None
    roles: list[str] = None

    def __post_init__(self):
        if self.roles is None:
            self.roles = []


_current_user: ContextVar[Optional[User]] = ContextVar('current_user', default=None)


class UserProvider:
    """
    Singleton for request-scoped user context.

    Set once per request in middleware, accessible everywhere without
    passing user through function parameters.
    """
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def set_user(self, user: User) -> None:
        _current_user.set(user)

    def get_user(self) -> Optional[User]:
        return _current_user.get()

    def clear_user(self) -> None:
        _current_user.set(None)

    @property
    def current_user(self) -> Optional[User]:
        return self.get_user()


# Global singleton
user_provider = UserProvider()


def get_current_user() -> Optional[User]:
    """Convenience function to get current user."""
    return user_provider.get_user()


def require_user() -> User:
    """Get current user or raise if not authenticated."""
    user = user_provider.get_user()
    if user is None:
        raise ValueError("No authenticated user in context")
    return user
```

**Usage in middleware:**
```python
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    # Validate token, get user...
    user = await validate_token(request)
    if user:
        user_provider.set_user(user)
    try:
        response = await call_next(request)
        return response
    finally:
        user_provider.clear_user()
```

**Usage in any function:**
```python
from src.shared.context import get_current_user, require_user

async def some_business_logic():
    user = require_user()  # Raises if not authenticated
    # Use user.id, user.tenant_id, etc.
```

---

## 6. Application Entry Point

### Clean main.py

```python
# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from src.shared.config import get_settings
from src.shared.logging import setup_logging, get_logger
from src.domains.router import root_router

settings = get_settings()
setup_logging(settings.log_level)
logger = get_logger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application startup and shutdown."""
    logger.info("=" * 60)
    logger.info(f"Starting {settings.app_name} v{settings.app_version}")
    logger.info(f"Debug: {settings.debug}")
    logger.info("=" * 60)

    # Initialize resources (DB connections, caches, etc.)

    yield

    # Cleanup resources
    logger.info("Shutting down")


app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    docs_url="/docs",
    redoc_url=None,
    lifespan=lifespan,
    debug=settings.debug,
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_credentials=settings.cors_credentials,
    allow_methods=settings.cors_methods,
    allow_headers=settings.cors_headers,
)

# Include all routes from domain router
app.include_router(root_router)


# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error", "type": type(exc).__name__}
    )
```

### Root Router

```python
# src/domains/router.py
from fastapi import APIRouter
from src.domains.health.controller import health_controller
# from src.domains.users.controller import users_controller
# from src.domains.items.controller import items_controller

root_router = APIRouter()

# Include all domain routers
root_router.include_router(health_controller.router)
# root_router.include_router(users_controller.router, prefix="/users", tags=["Users"])
# root_router.include_router(items_controller.router, prefix="/items", tags=["Items"])
```

---

## 7. Dependency Management

### CRITICAL: Always Start with Latest Safe Versions

**When setting up a new project, ALWAYS search for the latest version of each package and verify it has no known CVEs.**

Do NOT copy version numbers from old projects or templates. Package versions in documentation become outdated quickly.

**Process for each dependency:**
1. Search: `"{package} pypi latest version {current_year}"`
2. Check PyPI directly: `https://pypi.org/project/{package}/`
3. Search: `"{package} CVE vulnerability {current_year}"`
4. Verify no unpatched CVEs affect the latest version
5. Pin to the verified latest safe version

**Example search queries:**
```
"fastapi pypi latest version 2026"
"starlette CVE vulnerability 2026"
"pydantic-ai pypi latest version 2026"
```

### requirements.txt (Production)

```
# Production Dependencies
# Minor version pinning (~=) allows patch updates for security fixes
# CVE check date: YYYY-MM-DD  <-- UPDATE THIS DATE
# CVE sources: PyPI, GitHub Advisories, Snyk, NVD

# Core - VERIFY LATEST VERSIONS BEFORE USE
fastapi~=0.128.0
starlette~=0.50.0                      # CVE-2025-62727, CVE-2025-54121 fixed
uvicorn[standard]~=0.40.0
pydantic~=2.12.4                       # CVE-2024-3772 (ReDoS) fixed in 2.4.0+
pydantic-settings~=2.12.0

# HTTP client
httpx~=0.28.1
aiofiles~=25.1.0

# Utilities
python-multipart~=0.0.21
python-dotenv~=1.2.1
```

### requirements-dev.txt (Development)

```
# Development Dependencies
# NOT included in production Docker image
# CVE check date: YYYY-MM-DD  <-- UPDATE THIS DATE

-r requirements.txt

# Testing - VERIFY LATEST VERSIONS BEFORE USE
pytest~=9.0.2
anyio~=4.12.1                          # Includes pytest-anyio plugin
pytest-cov~=7.0.0

# Security auditing
pip-audit~=2.9.0

# Type checking
mypy~=1.19.1

# Linting (optional)
# ruff~=0.9.0
```

### CVE Check Process

Before adding or updating dependencies:

1. **Check PyPI** for security advisories: `https://pypi.org/project/{package}/`
2. **GitHub Security Advisories**: `https://github.com/advisories`
3. **Snyk vulnerability database**: `https://snyk.io/vuln`
4. **NVD**: `https://nvd.nist.gov/vuln/search`
5. **Run pip-audit**: `pip-audit` before releases

Document decisions in requirements.txt:
```
package~=1.2.0  # CVE-YYYY-XXXXX: pinned due to vulnerability in < 1.2.0
```

---

## 8. Exception Handling

### Custom Exception Hierarchy

```python
# src/shared/exceptions.py
from typing import Any, Optional


class AppException(Exception):
    """Base exception for application errors."""

    def __init__(
        self,
        message: str,
        status_code: int = 500,
        error_code: Optional[str] = None,
        details: Optional[dict[str, Any]] = None,
    ):
        self.message = message
        self.status_code = status_code
        self.error_code = error_code or self.__class__.__name__
        self.details = details or {}
        super().__init__(message)

    def to_dict(self) -> dict[str, Any]:
        return {
            "error": self.error_code,
            "message": self.message,
            "details": self.details,
        }


class NotFoundError(AppException):
    def __init__(self, resource: str, identifier: Any):
        super().__init__(
            message=f"{resource} not found: {identifier}",
            status_code=404,
            details={"resource": resource, "identifier": str(identifier)},
        )


class ValidationError(AppException):
    def __init__(self, message: str, field: Optional[str] = None):
        super().__init__(
            message=message,
            status_code=422,
            details={"field": field} if field else {},
        )


class AuthenticationError(AppException):
    def __init__(self, message: str = "Authentication required"):
        super().__init__(message=message, status_code=401)


class AuthorizationError(AppException):
    def __init__(self, message: str = "Permission denied"):
        super().__init__(message=message, status_code=403)


class ConflictError(AppException):
    def __init__(self, message: str):
        super().__init__(message=message, status_code=409)


class RateLimitError(AppException):
    def __init__(self, retry_after: int = 60):
        super().__init__(
            message="Rate limit exceeded",
            status_code=429,
            details={"retry_after": retry_after},
        )
```

### Exception Handler

```python
# In main.py
from src.shared.exceptions import AppException

@app.exception_handler(AppException)
async def app_exception_handler(request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content=exc.to_dict(),
    )
```

---

## 9. Testing

### conftest.py

```python
# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app


@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
async def client():
    """Async test client."""
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac
```

### Example Test

```python
# tests/test_health.py
import pytest


@pytest.mark.anyio
async def test_health_check(client):
    response = await client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"
```

---

## 10. Summary Checklist

### Project Setup
- [ ] Domain-based directory structure (`src/domains/`)
- [ ] Shared utilities in `src/shared/`
- [ ] Clean `main.py` (no routes, only app setup)
- [ ] Root router composing domain routers

### Patterns
- [ ] BaseController with lazy router
- [ ] BaseSchema with standard config
- [ ] Pydantic Settings for configuration
- [ ] Logger decorator with temporal benchmarking
- [ ] UserProvider singleton for request context
- [ ] Custom exception hierarchy

### Dependencies
- [ ] Minor version pinning (`~=`)
- [ ] Separate prod/dev requirements
- [ ] CVE check before updates
- [ ] pip-audit in CI/CD

### Quality
- [ ] Pytest with async support
- [ ] Type hints throughout
- [ ] Docstrings on public APIs
- [ ] CHANGELOG.md maintained