FastAPI Patterns

Build production-ready FastAPI applications with clean architecture, dependency injection, multi-tenancy support, and async patterns.

When to Use This Skill

• Creating new FastAPI applications or routers
• Implementing middleware (tenant resolution, logging, CORS)
• Designing dependency injection chains
• Integrating Supabase Auth with FastAPI
• Building multi-tenant API endpoints
• Configuring database sessions and connection pooling
• Structuring FastAPI projects for the monorepo

Project Structure

src/{app_name}/
├── __init__.py
├── main.py               # App factory and lifespan
├── config.py             # Settings (Pydantic BaseSettings)
├── dependencies.py       # Shared dependencies (auth, tenant, db)
├── middleware/
│   ├── __init__.py
│   ├── tenant.py         # Tenant resolution middleware
│   ├── logging.py        # Request logging middleware
│   └── timing.py         # Request timing middleware
├── routers/
│   ├── __init__.py
│   ├── invoices.py       # Invoice endpoints
│   ├── customers.py      # Customer endpoints
│   └── health.py         # Health check endpoints
├── models/               # SQLAlchemy models (if needed)
│   ├── __init__.py
│   └── invoice.py
├── schemas/              # Pydantic request/response schemas
│   ├── __init__.py
│   ├── invoice.py
│   ├── customer.py
│   └── common.py         # Shared schemas (pagination, errors)
├── services/             # Business logic layer
│   ├── __init__.py
│   ├── invoice_service.py
│   └── customer_service.py
└── utils/
    ├── __init__.py
    └── helpers.py

Application Factory Pattern

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

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

from app_name.config import Settings
from app_name.routers import invoices, customers, health
from app_name.middleware.tenant import TenantMiddleware
from app_name.middleware.logging import RequestLoggingMiddleware


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Application lifespan: startup and shutdown events."""
    # Startup
    settings = app.state.settings
    # Initialize database pool, caches, etc.
    yield
    # Shutdown
    # Close connections, flush buffers, etc.


def create_app(settings: Settings | None = None) -> FastAPI:
    """Application factory for FastAPI."""
    if settings is None:
        settings = Settings()

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

    # Store settings in app state
    app.state.settings = settings

    # Middleware (order matters: last added = first executed)
    app.add_middleware(RequestLoggingMiddleware)
    app.add_middleware(TenantMiddleware)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Routers
    app.include_router(health.router, tags=["health"])
    app.include_router(
        invoices.router,
        prefix="/v1/invoices",
        tags=["invoices"],
    )
    app.include_router(
        customers.router,
        prefix="/v1/customers",
        tags=["customers"],
    )

    return app

Configuration with Pydantic BaseSettings

from pydantic import Field
from pydantic_settings import BaseSettings


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

    model_config = {"env_prefix": "APP_", "env_file": ".env"}

    # Application
    app_name: str = "VeriFactu API"
    app_version: str = "0.1.0"
    debug: bool = False

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

    # Supabase
    supabase_url: str = Field(..., description="Supabase project URL")
    supabase_key: str = Field(..., description="Supabase anon key")
    supabase_jwt_secret: str = Field(..., description="JWT secret for validation")

    # Database
    database_url: str = Field(..., description="PostgreSQL connection string")
    db_pool_size: int = 5
    db_max_overflow: int = 10

    # CORS
    cors_origins: list[str] = ["http://localhost:5173"]

Dependency Injection

Authentication Dependency

from fastapi import Depends, HTTPException, Request, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import jwt, JWTError
from pydantic import BaseModel

security = HTTPBearer()


class AuthUser(BaseModel):
    """Authenticated user from JWT."""

    id: str
    email: str
    tenant_id: str | None = None
    role: str = "user"


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    request: Request = ...,
) -> AuthUser:
    """Validate Supabase JWT and extract user info."""
    settings = request.app.state.settings
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.supabase_jwt_secret,
            algorithms=["HS256"],
            audience="authenticated",
        )
        return AuthUser(
            id=payload["sub"],
            email=payload.get("email", ""),
            tenant_id=payload.get("tenant_id"),
            role=payload.get("role", "user"),
        )
    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
        ) from e


async def require_admin(
    user: AuthUser = Depends(get_current_user),
) -> AuthUser:
    """Require admin role."""
    if user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return user

Tenant Resolution Dependency

from fastapi import Depends, HTTPException, Request


class TenantContext(BaseModel):
    """Current tenant context."""

    tenant_id: str
    schema_name: str


async def get_current_tenant(
    request: Request,
    user: AuthUser = Depends(get_current_user),
) -> TenantContext:
    """Resolve tenant from authenticated user."""
    tenant_id = user.tenant_id
    if not tenant_id:
        # Fallback: check header
        tenant_id = request.headers.get("X-Tenant-ID")

    if not tenant_id:
        raise HTTPException(
            status_code=400,
            detail="Tenant context required",
        )

    return TenantContext(
        tenant_id=tenant_id,
        schema_name=f"tenant_{tenant_id}",
    )

Database Session Dependency

from collections.abc import AsyncIterator

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from fastapi import Depends, Request


def get_engine(request: Request) -> AsyncEngine:
    """Get the database engine from app state."""
    return request.app.state.db_engine


async def get_db(
    request: Request,
    tenant: TenantContext = Depends(get_current_tenant),
) -> AsyncIterator[AsyncSession]:
    """Provide a database session with tenant schema set."""
    engine = request.app.state.db_engine
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as session:
        # Set the search_path for tenant isolation
        await session.execute(
            text(f"SET search_path TO {tenant.schema_name}, public")
        )
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Router Patterns

CRUD Router

from fastapi import APIRouter, Depends, HTTPException, Query, status
from uuid import UUID

from app_name.dependencies import AuthUser, get_current_user, get_db
from app_name.schemas.invoice import (
    InvoiceCreate,
    InvoiceResponse,
    InvoiceUpdate,
    InvoiceListResponse,
)
from app_name.services.invoice_service import InvoiceService

router = APIRouter()


@router.get("", response_model=InvoiceListResponse)
async def list_invoices(
    page: int = Query(1, ge=1, description="Page number"),
    page_size: int = Query(20, ge=1, le=100, description="Items per page"),
    status_filter: str | None = Query(None, alias="status"),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> InvoiceListResponse:
    """List invoices with pagination and filtering."""
    service = InvoiceService(db)
    return await service.list_invoices(
        page=page,
        page_size=page_size,
        status_filter=status_filter,
    )


@router.post("", response_model=InvoiceResponse, status_code=status.HTTP_201_CREATED)
async def create_invoice(
    data: InvoiceCreate,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> InvoiceResponse:
    """Create a new invoice."""
    service = InvoiceService(db)
    return await service.create_invoice(data, created_by=user.id)


@router.get("/{invoice_id}", response_model=InvoiceResponse)
async def get_invoice(
    invoice_id: UUID,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> InvoiceResponse:
    """Get an invoice by ID."""
    service = InvoiceService(db)
    invoice = await service.get_invoice(invoice_id)
    if not invoice:
        raise HTTPException(status_code=404, detail="Invoice not found")
    return invoice


@router.patch("/{invoice_id}", response_model=InvoiceResponse)
async def update_invoice(
    invoice_id: UUID,
    data: InvoiceUpdate,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> InvoiceResponse:
    """Update an invoice (partial update)."""
    service = InvoiceService(db)
    invoice = await service.update_invoice(invoice_id, data)
    if not invoice:
        raise HTTPException(status_code=404, detail="Invoice not found")
    return invoice


@router.delete("/{invoice_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_invoice(
    invoice_id: UUID,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> None:
    """Delete an invoice."""
    service = InvoiceService(db)
    deleted = await service.delete_invoice(invoice_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Invoice not found")

Action Endpoints

@router.post("/{invoice_id}/submit", response_model=InvoiceResponse)
async def submit_invoice(
    invoice_id: UUID,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> InvoiceResponse:
    """Submit an invoice to the tax authority (VeriFactu)."""
    service = InvoiceService(db)
    return await service.submit_to_verifactu(invoice_id, submitted_by=user.id)

Middleware Patterns

Tenant Resolution Middleware

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


class TenantMiddleware(BaseHTTPMiddleware):
    """Extract tenant context from request and store in state."""

    SKIP_PATHS = {"/docs", "/redoc", "/openapi.json", "/health"}

    async def dispatch(self, request: Request, call_next) -> Response:
        if request.url.path in self.SKIP_PATHS:
            return await call_next(request)

        # Extract tenant from subdomain
        host = request.headers.get("host", "")
        parts = host.split(".")
        if len(parts) >= 3:
            request.state.tenant_slug = parts[0]
        else:
            request.state.tenant_slug = request.headers.get("X-Tenant-ID")

        return await call_next(request)

Request Logging Middleware

import time
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

logger = logging.getLogger(__name__)


class RequestLoggingMiddleware(BaseHTTPMiddleware):
    """Log all requests with timing information."""

    async def dispatch(self, request: Request, call_next) -> Response:
        start = time.perf_counter()
        response = await call_next(request)
        duration_ms = (time.perf_counter() - start) * 1000

        logger.info(
            "request completed",
            extra={
                "method": request.method,
                "path": request.url.path,
                "status": response.status_code,
                "duration_ms": round(duration_ms, 2),
                "tenant": getattr(request.state, "tenant_slug", None),
            },
        )

        response.headers["X-Request-Duration-Ms"] = str(round(duration_ms, 2))
        return response

Schema Patterns (Pydantic)

Request/Response Schemas

from datetime import datetime
from decimal import Decimal
from uuid import UUID

from pydantic import BaseModel, Field


class InvoiceCreate(BaseModel):
    """Schema for creating an invoice."""

    customer_id: UUID
    invoice_number: str = Field(..., min_length=1, max_length=50)
    amount: Decimal = Field(..., gt=0, decimal_places=2)
    currency: str = Field(default="EUR", pattern=r"^[A-Z]{3}$")
    notes: str | None = Field(None, max_length=1000)


class InvoiceUpdate(BaseModel):
    """Schema for partially updating an invoice."""

    amount: Decimal | None = Field(None, gt=0, decimal_places=2)
    notes: str | None = Field(None, max_length=1000)
    status: str | None = Field(None, pattern=r"^(draft|sent|paid|cancelled)$")


class InvoiceResponse(BaseModel):
    """Schema for invoice response."""

    id: UUID
    invoice_number: str
    customer_id: UUID
    amount: Decimal
    currency: str
    status: str
    created_at: datetime
    updated_at: datetime


class PaginationMeta(BaseModel):
    """Pagination metadata."""

    page: int
    page_size: int
    total_items: int
    total_pages: int
    has_next: bool
    has_prev: bool


class InvoiceListResponse(BaseModel):
    """Paginated list of invoices."""

    data: list[InvoiceResponse]
    pagination: PaginationMeta

Error Responses (RFC 7807)

from fastapi import Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel


class ProblemDetail(BaseModel):
    """RFC 7807 Problem Details response."""

    type: str = "about:blank"
    title: str
    status: int
    detail: str | None = None
    instance: str | None = None


async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse:
    """Convert HTTPExceptions to RFC 7807 format."""
    return JSONResponse(
        status_code=exc.status_code,
        content=ProblemDetail(
            title=exc.detail if isinstance(exc.detail, str) else "Error",
            status=exc.status_code,
            instance=str(request.url.path),
        ).model_dump(),
    )

Background Tasks

from fastapi import BackgroundTasks


@router.post("/{invoice_id}/send")
async def send_invoice(
    invoice_id: UUID,
    background_tasks: BackgroundTasks,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> InvoiceResponse:
    """Send invoice via email (async background task)."""
    service = InvoiceService(db)
    invoice = await service.get_invoice(invoice_id)
    if not invoice:
        raise HTTPException(status_code=404, detail="Invoice not found")

    # Queue email sending as background task
    background_tasks.add_task(
        send_invoice_email,
        invoice_id=invoice_id,
        recipient_email=invoice.customer_email,
    )

    return invoice


async def send_invoice_email(invoice_id: UUID, recipient_email: str) -> None:
    """Background task to send invoice email."""
    # Email sending logic
    ...

Pagination (Offset + Cursor)

Offset-Based Pagination

from sqlalchemy import select, func


async def paginate_query(
    session: AsyncSession,
    query,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list, PaginationMeta]:
    """Apply offset-based pagination to a query."""
    # Count total items
    count_query = select(func.count()).select_from(query.subquery())
    total_items = (await session.execute(count_query)).scalar_one()

    # Apply pagination
    offset = (page - 1) * page_size
    results = (
        await session.execute(query.offset(offset).limit(page_size))
    ).scalars().all()

    total_pages = (total_items + page_size - 1) // page_size

    meta = PaginationMeta(
        page=page,
        page_size=page_size,
        total_items=total_items,
        total_pages=total_pages,
        has_next=page < total_pages,
        has_prev=page > 1,
    )

    return list(results), meta

Cursor-Based Pagination

from datetime import datetime
from pydantic import BaseModel
import base64


class CursorPage(BaseModel):
    """Cursor-based pagination response."""

    data: list
    next_cursor: str | None = None
    has_more: bool = False


def encode_cursor(created_at: datetime, id: str) -> str:
    """Encode a cursor from sort key + ID."""
    raw = f"{created_at.isoformat()}|{id}"
    return base64.urlsafe_b64encode(raw.encode()).decode()


def decode_cursor(cursor: str) -> tuple[datetime, str]:
    """Decode a cursor into sort key + ID."""
    raw = base64.urlsafe_b64decode(cursor.encode()).decode()
    created_at_str, id_ = raw.split("|", 1)
    return datetime.fromisoformat(created_at_str), id_

Health Check Endpoint

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter()


class HealthResponse(BaseModel):
    status: str = "ok"
    version: str
    environment: str


@router.get("/health", response_model=HealthResponse)
async def health_check(request: Request) -> HealthResponse:
    """Health check endpoint for Kubernetes probes."""
    settings = request.app.state.settings
    return HealthResponse(
        version=settings.app_version,
        environment="production" if not settings.debug else "development",
    )

Testing Patterns

Test Client Setup

import pytest
from httpx import ASGITransport, AsyncClient

from app_name.main import create_app
from app_name.config import Settings


@pytest.fixture
def test_settings() -> Settings:
    return Settings(
        debug=True,
        supabase_url="http://localhost:54321",
        supabase_key="test-key",
        supabase_jwt_secret="test-secret",
        database_url="sqlite+aiosqlite:///test.db",
    )


@pytest.fixture
async def client(test_settings: Settings) -> AsyncClient:
    app = create_app(test_settings)
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.fixture
def auth_headers() -> dict[str, str]:
    """Create a valid JWT for testing."""
    from jose import jwt

    token = jwt.encode(
        {"sub": "user-123", "email": "test@example.com", "tenant_id": "acme"},
        "test-secret",
        algorithm="HS256",
    )
    return {"Authorization": f"Bearer {token}"}

Endpoint Tests

import pytest


@pytest.mark.anyio
async def test_list_invoices(client: AsyncClient, auth_headers: dict) -> None:
    response = await client.get("/v1/invoices", headers=auth_headers)
    assert response.status_code == 200
    data = response.json()
    assert "data" in data
    assert "pagination" in data


@pytest.mark.anyio
async def test_create_invoice(client: AsyncClient, auth_headers: dict) -> None:
    payload = {
        "customer_id": "550e8400-e29b-41d4-a716-446655440000",
        "invoice_number": "INV-2026-0001",
        "amount": "1500.00",
        "currency": "EUR",
    }
    response = await client.post(
        "/v1/invoices",
        json=payload,
        headers=auth_headers,
    )
    assert response.status_code == 201


@pytest.mark.anyio
async def test_unauthenticated_request(client: AsyncClient) -> None:
    response = await client.get("/v1/invoices")
    assert response.status_code == 403

API-First Design

• Design the API contract before implementing code
• Write OpenAPI 3.1 specifications as the source of truth
• Use the spec to generate client SDKs, documentation, and tests
• Keep the spec in version control alongside the code
• Review API design with stakeholders before implementation

URL Convention

# Resources (plural nouns)
GET    /v1/invoices                    # List invoices
POST   /v1/invoices                    # Create invoice
GET    /v1/invoices/{invoice_id}       # Get invoice
PATCH  /v1/invoices/{invoice_id}       # Update invoice
DELETE /v1/invoices/{invoice_id}       # Delete invoice

# Nested resources (ownership)
GET    /v1/customers/{customer_id}/invoices    # Customer's invoices

# Actions (when REST verbs don't fit)
POST   /v1/invoices/{invoice_id}/submit        # Submit to tax authority
POST   /v1/invoices/{invoice_id}/cancel        # Cancel invoice

# Query parameters
GET    /v1/invoices?status=draft&sort=-created_at&page=1&page_size=20

Versioning Strategy

• Use URL path versioning (/v1/invoices) for major breaking changes
• Use additive changes (new fields, new endpoints) for non-breaking evolution
• Deprecate before removing — communicate via response headers
• Maintain max 2 active API versions simultaneously
• Document migration guides between versions

Response Envelopes

Single Resource

{
  "data": {
    "id": "inv_abc123",
    "type": "invoice",
    "attributes": {
      "invoice_number": "INV-2026-0001",
      "amount": 1500.00,
      "currency": "EUR",
      "status": "draft"
    }
  }
}

Collection with Pagination

{
  "data": [
    { "id": "inv_abc123", "type": "invoice", "attributes": { ... } }
  ],
  "pagination": {
    "page": 1,
    "page_size": 20,
    "total_items": 47,
    "total_pages": 3,
    "has_next": true,
    "has_prev": false
  }
}

Rate Limiting

from fastapi import Request, HTTPException
from collections import defaultdict
from time import time

class RateLimiter:
    """Simple in-memory rate limiter per tenant."""

    def __init__(self, requests_per_minute: int = 60):
        self.rpm = requests_per_minute
        self.requests: dict[str, list[float]] = defaultdict(list)

    async def check(self, request: Request) -> None:
        tenant_id = getattr(request.state, "tenant_id", "anonymous")
        now = time()

        # Clean old entries
        self.requests[tenant_id] = [
            t for t in self.requests[tenant_id] if now - t < 60
        ]

        if len(self.requests[tenant_id]) >= self.rpm:
            raise HTTPException(
                status_code=429,
                detail="Rate limit exceeded",
                headers={"Retry-After": "60"},
            )

        self.requests[tenant_id].append(now)

OpenAPI 3.1 Specification Pattern

openapi: "3.1.0"
info:
  title: EasyFactu API
  version: "1.0.0"
  description: REST API for invoice management and Spanish tax compliance.

servers:
  - url: https://api.easyfactu.es/v1
    description: Production
  - url: http://localhost:8000/v1
    description: Local development

security:
  - BearerAuth: []

paths:
  /invoices:
    get:
      operationId: listInvoices
      summary: List invoices
      tags: [Invoices]
      parameters:
        - name: page
          in: query
          schema: { type: integer, minimum: 1, default: 1 }
        - name: page_size
          in: query
          schema: { type: integer, minimum: 1, maximum: 100, default: 20 }
        - name: status
          in: query
          schema: { $ref: "#/components/schemas/InvoiceStatus" }
      responses:
        "200":
          description: Paginated list of invoices
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/PaginatedInvoices"
        "401":
          $ref: "#/components/responses/Unauthorized"

components:
  securitySchemes:
    BearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT

  schemas:
    InvoiceStatus:
      type: string
      enum: [draft, sent, paid, cancelled]

  responses:
    Unauthorized:
      description: Authentication required
      content:
        application/json:
          schema:
            $ref: "#/components/schemas/ProblemDetail"

Guidelines

• Always use the application factory pattern for testability
• Keep routers thin — delegate business logic to services
• Use Pydantic models for all request/response schemas
• Apply dependency injection for cross-cutting concerns (auth, db, tenant)
• Use async/await for all I/O-bound operations
• Return RFC 7807 error responses for consistency
• Include health check endpoints for Kubernetes probes
• Write integration tests using httpx.AsyncClient with ASGITransport
• Design the API contract before writing code (API-first)
• Use URL path versioning (/v1/) for breaking changes
• Apply rate limiting per tenant