Backend Patterns

Backend Patterns — FastAPI

Production-grade FastAPI patterns for Python backends: system design with OOP layering, dependency injection, error contracts, background processing, real-time WebSocket, async database integration, and comprehensive testing.

When to Use

• Building or refactoring a FastAPI service from scratch or adding endpoints to an existing app.
• Designing layered architecture with entities, services, repositories, and domain events.
• Implementing dependency injection, middleware pipelines, or custom error handlers.
• Adding background task processing (celery, RQ, or FastAPI BackgroundTasks).
• Creating real-time WebSocket endpoints with connection lifecycle management.
• Integrating SQLAlchemy async ORM with Alembic migration workflows.
• Designing OpenAPI-compliant APIs with envelopes, versioning, and reusable schemas.
• Writing comprehensive async tests with TestClient and fixture patterns.

Core Concepts

1. System Design & OOP Layering

Adopt a clean layered architecture that separates concerns across four tiers:

┌─────────────────────────────────────────────┐
│  Presentation Layer (FastAPI routers)        │
│  - Route handlers, request/response models   │
│  - Dependency injection wiring              │
├─────────────────────────────────────────────┤
│  Service Layer (business logic)              │
│  - Use cases, orchestration, domain events  │
│  - No direct HTTP or DB code                │
├─────────────────────────────────────────────┤
│  Repository Layer (data access)              │
│  - CRUD operations, query builders           │
│  - Returns domain entities, not ORM objects  │
├─────────────────────────────────────────────┤
│  Domain Layer (entities, value objects)      │
│  - Pure Python, no framework dependencies    │
│  - Domain events, invariants, value objects  │
└─────────────────────────────────────────────┘

Domain Entities — Pure Python dataclasses or Pydantic models with no framework coupling:

from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID, uuid4

@dataclass
class User:
    id: UUID = field(default_factory=uuid4)
    email: str = ""
    name: str = ""
    is_active: bool = True
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)

    def deactivate(self) -> None:
        if not self.is_active:
            raise ValueError("User is already inactive")
        self.is_active = False
        self.updated_at = datetime.utcnow()

Domain Events — Lightweight event objects published by services:

from dataclasses import dataclass
from datetime import datetime

@dataclass
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.utcnow)

@dataclass
class UserCreated(DomainEvent):
    user_id: UUID
    email: str

@dataclass
class UserDeactivated(DomainEvent):
    user_id: UUID

Repository Interface — Abstract contract for data access:

from abc import ABC, abstractmethod
from typing import Optional
from uuid import UUID

from app.domain.entities import User

class UserRepository(ABC):
    @abstractmethod
    async def get_by_id(self, user_id: UUID) -> Optional[User]: ...

    @abstractmethod
    async def get_by_email(self, email: str) -> Optional[User]: ...

    @abstractmethod
    async def create(self, user: User) -> User: ...

    @abstractmethod
    async def update(self, user: User) -> User: ...

    @abstractmethod
    async def list_all(self, skip: int = 0, limit: int = 50) -> list[User]: ...

Service Layer — Orchestrates use cases, publishes domain events:

from typing import Protocol
from uuid import UUID

from app.domain.entities import User
from app.domain.events import UserCreated, UserDeactivated
from app.repositories import UserRepository

class EventPublisher(Protocol):
    async def publish(self, event: DomainEvent) -> None: ...

class UserService:
    def __init__(
        self,
        repo: UserRepository,
        publisher: EventPublisher,
    ) -> None:
        self.repo = repo
        self.publisher = publisher

    async def create_user(self, email: str, name: str) -> User:
        existing = await self.repo.get_by_email(email)
        if existing:
            raise DuplicateResourceError(f"User with email {email} already exists")

        user = User(email=email, name=name)
        user = await self.repo.create(user)
        await self.publisher.publish(UserCreated(user_id=user.id, email=user.email))
        return user

    async def deactivate_user(self, user_id: UUID) -> User:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise ResourceNotFoundError(f"User {user_id} not found")

        user.deactivate()
        user = await self.repo.update(user)
        await self.publisher.publish(UserDeactivated(user_id=user.id))
        return user

2. FastAPI App Structure

Organize the application with routers, dependencies, middleware, and lifespan:

app/
├── __init__.py
├── main.py                # create_app(), lifespan, middleware wiring
├── config.py              # pydantic-settings BaseSettings
├── dependencies.py        # reusable Depends() providers
├── middleware.py           # custom middleware factories
├── exceptions.py          # custom exceptions + exception handlers
├── domain/
│   ├── entities.py
│   ├── events.py
│   └── errors.py
├── repositories/
│   ├── base.py            # Abstract base repository
│   └── user_repository.py
├── services/
│   ├── user_service.py
│   └── event_publisher.py
├── routers/
│   ├── users.py
│   └── health.py
├── schemas/
│   ├── user.py            # Pydantic request/response models
│   └── common.py          # Envelope, pagination, error schemas
├── database.py            # async engine, session factory
└── middleware/
    ├── auth.py
    ├── cors.py
    └── rate_limit.py

Application Factory with Lifespan:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.config import Settings
from app.database import init_db, close_db
from app.exceptions import register_exception_handlers
from app.middleware.cors import setup_cors
from app.middleware.rate_limit import RateLimitMiddleware
from app.routers import users, health

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Startup: initialize database, warm caches
    await init_db()
    yield
    # Shutdown: close connections, flush buffers
    await close_db()

def create_app(settings: Settings | None = None) -> FastAPI:
    app = FastAPI(
        title="AI Dev Kit API",
        version="1.0.0",
        lifespan=lifespan,
        docs_url="/docs",
        openapi_url="/openapi.json",
    )

    # Middleware
    setup_cors(app, origins=settings.allowed_origins if settings else [])
    app.add_middleware(RateLimitMiddleware, requests_per_minute=60)

    # Exception handlers
    register_exception_handlers(app)

    # Routers
    app.include_router(users.router, prefix="/api/v1")
    app.include_router(health.router, prefix="/api/v1")

    return app

3. Pydantic v2 Models

Use Pydantic v2's BaseModel, Field, field_validator, and model_config:

from __future__ import annotations

from datetime import datetime
from typing import Optional
from uuid import UUID

from pydantic import (
    BaseModel,
    EmailStr,
    Field,
    field_validator,
    model_validator,
    ConfigDict,
)

class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=150)
    password: str = Field(..., min_length=8, max_length=128)

    model_config = ConfigDict(
        str_strip_whitespace=True,
        json_schema_extra={
            "examples": [
                {
                    "email": "user @example.com",
                    "name": "Jane Doe",
                    "password": "s3cur3P@ss!",
                }
            ]
        },
    )

    @field_validator("name")
    @classmethod
    def name_must_not_be_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name must not be blank")
        return v.strip()

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain at least one uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain at least one digit")
        return v

class UserResponse(BaseModel):
    id: UUID
    email: EmailStr
    name: str
    is_active: bool
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

4. Dependency Injection

Use Depends() for reusable, composable providers:

from collections.abc import AsyncIterator
from typing import Annotated

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

from app.database import async_session_factory
from app.domain.entities import User
from app.repositories.user_repository import SQLAlchemyUserRepository
from app.services.user_service import UserService
from app.services.event_publisher import RedisEventPublisher

# --- Database session ---
async def get_db_session() -> AsyncIterator[AsyncSession]:
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# --- Repository ---
def get_user_repo(
    session: Annotated[AsyncSession, Depends(get_db_session)],
) -> SQLAlchemyUserRepository:
    return SQLAlchemyUserRepository(session)

# --- Service ---
def get_user_service(
    repo: Annotated[SQLAlchemyUserRepository, Depends(get_user_repo)],
) -> UserService:
    publisher = RedisEventPublisher()
    return UserService(repo=repo, publisher=publisher)

# --- Auth ---
async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    service: Annotated[UserService, Depends(get_user_service)],
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    # decode JWT, look up user, raise if invalid
    user = await authenticate_token(token, service)
    if user is None:
        raise credentials_exception
    if not user.is_active:
        raise HTTPException(status_code=403, detail="Inactive user")
    return user

# Type aliases for ergonomic injection
DBSession = Annotated[AsyncSession, Depends(get_db_session)]
UserServiceDep = Annotated[UserService, Depends(get_user_service)]
CurrentUser = Annotated[User, Depends(get_current_user)]

5. Error Handling

Define a unified error contract with custom exceptions and handlers:

# app/exceptions.py
from __future__ import annotations

from http import HTTPStatus
from typing import Any

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

class AppError(Exception):
    """Base application error with HTTP status and user-safe detail."""

    status_code: int = HTTPStatus.INTERNAL_SERVER_ERROR
    detail: str = "Internal server error"
    extra: dict[str, Any] | None = None

class ResourceNotFoundError(AppError):
    status_code = HTTPStatus.NOT_FOUND
    detail = "Resource not found"

class DuplicateResourceError(AppError):
    status_code = HTTPStatus.CONFLICT
    detail = "Resource already exists"

class UnauthorizedError(AppError):
    status_code = HTTPStatus.UNAUTHORIZED
    detail = "Authentication required"

class ValidationError(AppError):
    status_code = HTTPStatus.UNPROCESSABLE_ENTITY
    detail = "Validation failed"

class ErrorEnvelope(BaseModel):
    success: bool = False
    error: dict[str, Any]

def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "success": False,
                "error": {
                    "code": exc.status_code,
                    "message": exc.detail,
                    **(exc.extra or {}),
                },
            },
        )

    @app.exception_handler(Exception)
    async def unhandled_error(request: Request, exc: Exception) -> JSONResponse:
        # Log stack trace in production; never leak internals to client
        app.logger.exception("Unhandled exception: %s", exc)
        return JSONResponse(
            status_code=500,
            content={
                "success": False,
                "error": {"code": 500, "message": "Internal server error"},
            },
        )

Error Contract in Schemas:

# app/schemas/common.py
from typing import Generic, TypeVar, Optional, Any
from pydantic import BaseModel

T = TypeVar("T")

class PaginationMeta(BaseModel):
    total: int
    page: int
    page_size: int
    pages: int

class APIEnvelope(BaseModel, Generic[T]):
    success: bool
    data: Optional[T] = None
    error: Optional[dict[str, Any]] = None
    pagination: Optional[PaginationMeta] = None

6. Background Tasks

FastAPI BackgroundTasks (lightweight):

from fastapi import BackgroundTasks, Depends
from app.schemas.user import UserCreate, UserResponse
from app.services.user_service import UserService

@router.post("/users", response_model=APIEnvelope[UserResponse], status_code=201)
async def create_user(
    payload: UserCreate,
    bg: BackgroundTasks,
    service: UserServiceDep,
) -> APIEnvelope[UserResponse]:
    user = await service.create_user(email=payload.email, name=payload.name)
    bg.add_task(send_welcome_email, user.email, user.name)
    return APIEnvelope(success=True, data=UserResponse.model_validate(user))

def send_welcome_email(email: str, name: str) -> None:
    """Runs after the response is returned — non-blocking."""
    ...

Celery Integration (heavy/long-running tasks):

# app/worker/celery_app.py
from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    worker_prefetch_multiplier=1,
)

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def generate_report(self, user_id: str, params: dict) -> dict:
    try:
        # Long-running report generation
        result = build_report(user_id, params)
        return result
    except Exception as exc:
        self.retry(exc=exc)

7. WebSocket — Real-Time Endpoints

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import Dict, Set
import asyncio
import json

router = APIRouter()

class ConnectionManager:
    """Manages WebSocket connections with room-based grouping."""

    def __init__(self) -> None:
        self.active_connections: Dict[str, Set[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room: str) -> None:
        await websocket.accept()
        self.active_connections.setdefault(room, set()).add(websocket)

    def disconnect(self, websocket: WebSocket, room: str) -> None:
        self.active_connections.get(room, set()).discard(websocket)
        if not self.active_connections.get(room):
            self.active_connections.pop(room, None)

    async def broadcast(self, room: str, message: dict) -> None:
        disconnected = set()
        for ws in self.active_connections.get(room, set()):
            try:
                await ws.send_json

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [noah-sheldon](https://github.com/noah-sheldon)
- **Source:** [noah-sheldon/ai-dev-kit](https://github.com/noah-sheldon/ai-dev-kit)
- **License:** MIT
- **Homepage:** https://noahsheldon.dev

Install and usage instructions live in the source repository linked above.