Ecc Fastapi Patterns

FastAPI Patterns

Production-oriented patterns for FastAPI services.

When to Use

• Building or reviewing a FastAPI app.
• Splitting routers, schemas, dependencies, and database access.
• Writing async endpoints that call a database or external service.
• Adding authentication, authorization, OpenAPI docs, tests, or deployment settings.
• Checking a FastAPI PR for copy-pasteable examples and production risks.

How It Works

Treat the FastAPI app as a thin HTTP layer over explicit dependencies and service code:

• main.py owns app construction, middleware, exception handlers, and router registration.
• schemas/ owns Pydantic request and response models.
• dependencies.py owns database, auth, pagination, and request-scoped dependencies.
• services/ or crud/ owns business and persistence operations.
• tests/ overrides dependencies instead of opening production resources.

Prefer small routers and explicit response_model declarations. Keep raw ORM objects, secrets, and framework globals out of response schemas.

Project Layout

app/
|-- main.py
|-- config.py
|-- dependencies.py
|-- exceptions.py
|-- api/
|   `-- routes/
|       |-- users.py
|       `-- health.py
|-- core/
|   |-- security.py
|   `-- middleware.py
|-- db/
|   |-- session.py
|   `-- crud.py
|-- models/
|-- schemas/
`-- tests/

Application Factory

Use a factory so tests and workers can build the app with controlled settings.

from contextlib import asynccontextmanager

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

from app.api.routes import health, users
from app.config import settings
from app.db.session import close_db, init_db
from app.exceptions import register_exception_handlers


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield
    await close_db()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.api_title,
        version=settings.api_version,
        lifespan=lifespan,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=bool(settings.cors_origins),
        allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
        allow_headers=["Authorization", "Content-Type"],
    )

    register_exception_handlers(app)
    app.include_router(health.router, prefix="/health", tags=["health"])
    app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
    return app


app = create_app()

Do not use allow_origins=["*"] with allow_credentials=True; browsers reject that combination and Starlette disallows it for credentialed requests.

Pydantic Schemas

Keep request, update, and response models separate.

from datetime import datetime
from typing import Annotated
from uuid import UUID

from pydantic import BaseModel, ConfigDict, EmailStr, Field


class UserBase(BaseModel):
    email: EmailStr
    full_name: Annotated[str, Field(min_length=1, max_length=100)]


class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=12, max_length=128)]


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: Annotated[str | None, Field(min_length=1, max_length=100)] = None


class UserResponse(UserBase):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    created_at: datetime
    updated_at: datetime

Response models must never include password hashes, access tokens, refresh tokens, or internal authorization state.

Dependencies

Use dependency injection for request-scoped resources.

from collections.abc import AsyncIterator
from uuid import UUID

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import decode_token
from app.db.session import session_factory
from app.models.user import User


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


async def get_db() -> AsyncIterator[AsyncSession]:
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    payload = decode_token(token)
    user_id = UUID(payload["sub"])
    user = await db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return user

Avoid creating sessions, clients, or credentials inline inside route handlers.

Async Endpoints

Keep route handlers async when they perform I/O, and use async libraries inside them.

from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_current_user, get_db
from app.models.user import User
from app.schemas.user import UserResponse


router = APIRouter()


@router.get("/", response_model=list[UserResponse])
async def list_users(
    limit: int = Query(default=50, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(
        select(User).order_by(User.created_at.desc()).limit(limit).offset(offset)
    )
    return result.scalars().all()

Use httpx.AsyncClient for external HTTP calls from async handlers. Do not call requests in an async route.

Error Handling

Centralize domain exceptions and keep response shapes stable.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class ApiError(Exception):
    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(ApiError)
    async def api_error_handler(request: Request, exc: ApiError):
        return JSONResponse(
            status_code=exc.status_code,
            content={"error": {"code": exc.code, "message": exc.message}},
        )

OpenAPI Customization

Assign the custom OpenAPI callable to app.openapi; do not just call the function once.

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi


def install_openapi(app: FastAPI) -> None:
    def custom_openapi():
        if app.openapi_schema:
            return app.openapi_schema
        app.openapi_schema = get_openapi(
            title="Service API",
            version="1.0.0",
            routes=app.routes,
        )
        return app.openapi_schema

    app.openapi = custom_openapi

Testing

Override the dependency used by Depends, not an internal helper that route handlers never reference.

import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.main import create_app


@pytest.fixture
async def client(test_session: AsyncSession):
    app = create_app()

    async def override_get_db():
        yield test_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as test_client:
        yield test_client
    app.dependency_overrides.clear()

Security Checklist

• Hash passwords with argon2-cffi, bcrypt, or a current passlib-compatible hasher.
• Validate JWT issuer, audience, expiry, and signing algorithm.
• Keep CORS origins environment-specific.
• Put rate limits on auth and write-heavy endpoints.
• Use Pydantic models for all request bodies.
• Use ORM parameter binding or SQLAlchemy Core expressions; never build SQL with f-strings.
• Redact tokens, authorization headers, cookies, and passwords from logs.
• Run dependency audit tooling in CI.

Performance Checklist

• Configure database connection pooling explicitly.
• Add pagination to list endpoints.
• Watch for N+1 queries and use eager loading intentionally.
• Use async HTTP/database clients in async paths.
• Add compression only after checking payload size and CPU tradeoffs.
• Cache stable expensive reads behind explicit invalidation.

Examples

Use these examples as patterns, not as project-wide templates:

• Application factory: configure middleware and routers once in create_app.
• Schema split: UserCreate, UserUpdate, and UserResponse have different responsibilities.
• Dependency override: tests override get_db directly.
• OpenAPI customization: assign app.openapi = custom_openapi.

See Also

• Agent: fastapi-reviewer
• Command: /fastapi-review
• Skill: python-patterns
• Skill: python-testing
• Skill: api-design