fastapi-patterns

name	fastapi-patterns
description	FastAPI architecture patterns, async SQLAlchemy ORM, Pydantic schemas, dependency injection, service layer, background tasks, and production-grade API design.
origin	local

FastAPI Development Patterns

Production-grade FastAPI architecture patterns with async SQLAlchemy and Pydantic v2.

When to Activate

• Building FastAPI applications
• Designing REST APIs with FastAPI + SQLAlchemy
• Working with async database sessions
• Setting up project structure and dependency injection
• Implementing service layers, background tasks, pagination

Project Structure

Recommended Layout

myproject/
├── app/
│   ├── __init__.py
│   ├── main.py              # App factory, lifespan, router registration
│   ├── config.py            # Settings via pydantic-settings
│   ├── database.py          # Async engine, session factory
│   ├── dependencies.py      # Shared Depends (db, current_user, etc.)
│   ├── models/              # SQLAlchemy ORM models
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── user.py
│   │   └── product.py
│   ├── schemas/             # Pydantic request/response models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   ├── routers/             # APIRouter per domain
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── products.py
│   └── services/            # Business logic
│       ├── __init__.py
│       ├── user_service.py
│       └── product_service.py
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── tests/
├── alembic.ini
├── pyproject.toml
└── .env

App Factory with Lifespan

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.database import engine
from app.routers import users, products


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    yield
    # Shutdown
    await engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        version=settings.app_version,
        docs_url="/docs" if settings.debug else None,
        redoc_url=None,
        lifespan=lifespan,
    )

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    app.include_router(users.router, prefix="/api/v1")
    app.include_router(products.router, prefix="/api/v1")

    return app


app = create_app()

Settings with pydantic-settings

# app/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    app_name: str = "MyApp"
    app_version: str = "0.1.0"
    debug: bool = False

    database_url: str
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    cors_origins: list[str] = ["http://localhost:3000"]


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


settings = get_settings()

Database Setup (Async SQLAlchemy)

# app/database.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from app.config import settings

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_pre_ping=True,
    pool_size=10,
    max_overflow=20,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
    autocommit=False,
)


class Base(DeclarativeBase):
    pass

Model Design Patterns

SQLAlchemy Async Models

# app/models/base.py
from datetime import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )


# app/models/user.py
from sqlalchemy import String, Boolean
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
from app.models.base import TimestampMixin


class User(Base, TimestampMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_superuser: Mapped[bool] = mapped_column(Boolean, default=False)

    products: Mapped[list[]] = relationship(back_populates=, lazy=)



 decimal  Decimal
 sqlalchemy  String, Text, Numeric, Integer, ForeignKey, Index, CheckConstraint
 sqlalchemy.orm  Mapped, mapped_column, relationship
 app.database  Base
 app.models.base  TimestampMixin


 (Base, TimestampMixin):
    __tablename__ = 
    __table_args__ = (
        Index(, , ),
        CheckConstraint(, name=),
        CheckConstraint(, name=),
    )

    : Mapped[] = mapped_column(primary_key=, index=)
    name: Mapped[] = mapped_column(String(), index=)
    description: Mapped[ | ] = mapped_column(Text)
    price: Mapped[Decimal] = mapped_column(Numeric(, ))
    stock: Mapped[] = mapped_column(Integer, default=)
    is_active: Mapped[] = mapped_column(default=)
    owner_id: Mapped[] = mapped_column(ForeignKey(), index=)

    owner: Mapped[] = relationship(back_populates=)

Pydantic Schema Patterns

# app/schemas/product.py
from decimal import Decimal
from datetime import datetime
from pydantic import BaseModel, Field, ConfigDict


class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=200)
    description: str | None = None
    price: Decimal = Field(..., ge=0, decimal_places=2)
    stock: int = Field(..., ge=0)


class ProductCreate(ProductBase):
    pass


class ProductUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=200)
    description: str | None = None
    price: Decimal | None = Field(None, ge=0, decimal_places=2)
    stock: int | None = Field(None, ge=0)
    is_active: bool | None = None


class ProductResponse(ProductBase):
    model_config = ConfigDict(from_attributes=)

    : 
    is_active: 
    owner_id: 
    created_at: datetime
    updated_at: datetime


 ():
    items: [ProductResponse]
    total: 
    page: 
    size: 
    pages: 

Dependency Injection

# app/dependencies.py
from typing import Annotated, AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import AsyncSessionLocal


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


DbSession = Annotated[AsyncSession, Depends(get_db)]

Router Patterns

# app/routers/products.py
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import DbSession
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse, ProductListResponse
from app.services.product_service import ProductService
from app.routers.users import CurrentUser

router = APIRouter(prefix="/products", tags=["products"])


@router.get("/", response_model=ProductListResponse)
async def list_products(
    db: DbSession,
    page: int = Query(1, ge=1),
    size: int = Query(20, ge=1, le=100),
    search: str | None = Query(None),
):
    service = ProductService(db)
    return await service.list(page=page, size=size, search=search)


@router.get("/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int, db: DbSession):
    service = ProductService(db)
    product =  service.get_or_404(product_id)
     product



  ():
    service = ProductService(db)
      service.create(data, owner_id=current_user.)



  ():
    service = ProductService(db)
    product =  service.get_or_404(product_id)
     product.owner_id != current_user.:
         HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=)
      service.update(product, data)



  ():
    service = ProductService(db)
    product =  service.get_or_404(product_id)
     product.owner_id != current_user.:
         HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=)
     service.delete(product)

Service Layer Pattern

# app/services/product_service.py
import math
from sqlalchemy import select, func, or_
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import HTTPException, status

from app.models.product import Product
from app.schemas.product import ProductCreate, ProductUpdate, ProductListResponse


class ProductService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def get_or_404(self, product_id: int) -> Product:
        result = await self.db.get(Product, product_id)
        if not result:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
        return result

    async def list(
        self,
        page: int = 1,
        size: int = 20,
        search: str | None = None,
    ) -> ProductListResponse:
        query = select(Product).where(Product.is_active == True)

        if search:
            query = query.where(
                or_(
                    Product.name.ilike(f"%%"),
                    Product.description.ilike(),
                )
            )

        total_result =  .db.execute(select(func.count()).select_from(query.subquery()))
        total = total_result.scalar_one()

        query = query.offset((page - ) * size).limit(size)
        result =  .db.execute(query)
        items = (result.scalars().())

         ProductListResponse(
            items=items,
            total=total,
            page=page,
            size=size,
            pages=math.ceil(total / size)  total  ,
        )

      () -> Product:
        product = Product(**data.model_dump(), owner_id=owner_id)
        .db.add(product)
         .db.flush()
         .db.refresh(product)
         product

      () -> Product:
         field, value  data.model_dump(exclude_unset=).items():
            (product, field, value)
         .db.flush()
         .db.refresh(product)
         product

      () -> :
         .db.delete(product)
         .db.flush()

Background Tasks

# app/routers/orders.py
from fastapi import APIRouter, BackgroundTasks
from app.services.email_service import send_order_confirmation

router = APIRouter(prefix="/orders", tags=["orders"])


@router.post("/", status_code=201)
async def create_order(
    data: OrderCreate,
    background_tasks: BackgroundTasks,
    db: DbSession,
    current_user: CurrentUser,
):
    service = OrderService(db)
    order = await service.create(data, user_id=current_user.id)
    background_tasks.add_task(send_order_confirmation, order.id, current_user.email)
    return order

Alembic Setup

# alembic/env.py
import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

from app.config import settings
from app.database import Base
# Import all models so Alembic can detect them
import app.models  # noqa: F401

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    context.configure(
        url=settings.database_url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect()  connection:
         connection.run_sync(do_run_migrations)
     connectable.dispose()


 () -> :
    asyncio.run(run_async_migrations())


 context.is_offline_mode():
    run_migrations_offline()
:
    run_migrations_online()

Error Handling

# app/main.py (add to create_app)
from fastapi import Request
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError


@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
    return JSONResponse(
        status_code=409,
        content={"detail": "Resource already exists or constraint violation"},
    )

Quick Reference

Pattern	Description
AsyncSession	Always use async DB session
Depends(get_db)	Inject DB session, auto-commit/rollback
Annotated[X, Depends(...)]	Reusable typed dependency aliases
mapped_column	Modern SQLAlchemy 2.x column definition
model_dump(exclude_unset=True)	Partial updates — only set fields
expire_on_commit=False	Keep objects accessible after commit
lazy="selectin"	Async-safe relationship loading
lru_cache on settings	Load env vars once
lifespan	Startup/shutdown events
BackgroundTasks	Fire-and-forget async work

Build for explicitness: FastAPI's DI system is powerful, but keep dependencies focused and services stateless (except for the db session).