FastAPI Architecture: Design Patterns and Project Structure

FastAPI Architecture: Design Patterns and Project Structure

What is FastAPI Architecture?

FastAPI architecture refers to the strategic organization of code, separation of concerns, and adoption of proven design patterns that transform a simple API into a maintainable, scalable, and testable application. While FastAPI itself is unopinionated about project structure, the framework's powerful dependency injection system, async capabilities, and type-driven design naturally lend themselves to layered architectures that keep business logic decoupled from transport concerns.

At its core, a well-architected FastAPI application typically follows a layered approach with clear boundaries between:

• Transport Layer — Routers, middleware, and request/response handling
• Service Layer — Business logic orchestration and use cases
• Repository Layer — Data access and persistence abstractions
• Domain Layer — Core entities, value objects, and Pydantic schemas

This separation allows each layer to evolve independently, makes unit testing straightforward, and prevents framework lock-in by keeping FastAPI-specific code confined to the transport layer.

Why Architecture Matters

Without deliberate architecture, FastAPI projects tend to accumulate all logic inside route functions — database queries, validation, business rules, and external API calls become tangled together. This leads to several pain points as the codebase grows:

• Testing becomes painful — Routes require HTTP test clients even when you only want to verify business logic
• Code duplication spreads — Similar queries and validations get copy-pasted across route files
• Onboarding slows dramatically — New developers must understand the entire stack to make a single change
• Database migrations trigger cascading changes — ORM models leak into route handlers and response schemas
• Async benefits are lost — Blocking operations mixed with async code create bottlenecks

Investing in architecture upfront — even for small projects — pays dividends within weeks as features accumulate and requirements shift. The patterns described here are lightweight enough for microservices yet scale elegantly to monoliths with dozens of modules.

Core Design Patterns for FastAPI

1. Router Separation Pattern

The most fundamental pattern is splitting routes by domain concern using FastAPI's APIRouter. Rather than a single monolithic app.py, each domain gets its own router file with a consistent prefix and tag set. This enables independent versioning, clean OpenAPI documentation grouping, and the ability to mount sub-applications.

# api/routers/users.py
from fastapi import APIRouter

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

@router.get("/")
async def list_users():
    ...

@router.post("/")
async def create_user():
    ...

2. Dependency Injection Chain Pattern

FastAPI's dependency injection is not just for request-level concerns like authentication — it forms the backbone of a clean architecture. Dependencies can themselves depend on other dependencies, creating a chain that wires up the entire application stack without any service locator or global state.

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncSession:
    # Provides a database session
    ...

async def get_user_repository(
    db: AsyncSession = Depends(get_db)
) -> UserRepository:
    return UserRepository(db)

async def get_user_service(
    repo: UserRepository = Depends(get_user_repository)
) -> UserService:
    return UserService(repo)

@router.post("/users")
async def create_user(
    service: UserService = Depends(get_user_service)
):
    return await service.create_user(...)

This chain means route functions never directly touch the database — they depend on a service, which depends on a repository, which depends on a session. Each link is easily mocked in tests.

3. Repository Pattern

The repository pattern abstracts data access behind a plain class interface. This is crucial even with ORMs because it consolidates query logic, prevents duplicate query fragments, and lets you swap storage backends (or mock entirely) without touching business logic.

# repositories/base.py
from typing import TypeVar, Generic, Optional, List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

T = TypeVar('T')

class BaseRepository(Generic[T]):
    def __init__(self, session: AsyncSession, model: type[T]):
        self.session = session
        self.model = model

    async def get_by_id(self, id: int) -> Optional[T]:
        result = await self.session.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalar_one_or_none()

    async def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
        result = await self.session.execute(
            select(self.model).offset(skip).limit(limit)
        )
        return result.scalars().all()

    async def create(self, entity: T) -> T:
        self.session.add(entity)
        await self.session.flush()
        await self.session.refresh(entity)
        return entity

    async def delete(self, entity: T) -> None:
        await self.session.delete(entity)
        await self.session.flush()

4. Service Layer Pattern

The service layer houses all business logic — validation rules, orchestration of multiple repositories, transaction boundaries, and integration with external services. Services are plain Python classes (or async functions) that receive repositories via dependency injection, never via global imports.

# services/user_service.py
from typing import Optional
from repositories.user_repository import UserRepository
from models.user import User
from schemas.user import UserCreate, UserUpdate
from core.exceptions import ConflictError, NotFoundError
from core.security import hash_password

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create_user(self, data: UserCreate) -> User:
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError("Email already registered")

        user = User(
            email=data.email,
            hashed_password=hash_password(data.password),
            full_name=data.full_name
        )
        return await self.repo.create(user)

    async def get_user(self, user_id: int) -> Optional[User]:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise NotFoundError("User not found")
        return user

    async def update_user(self, user_id: int, data: UserUpdate) -> User:
        user = await self.get_user(user_id)
        update_data = data.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(user, key, value)
        await self.repo.session.flush()
        return user

5. Model-Schema Separation Pattern

A critical distinction in FastAPI architecture is keeping ORM models (database tables) completely separate from Pydantic schemas (API contracts). Never return an ORM model directly from a route — always convert through a Pydantic schema. This prevents accidental exposure of internal fields, keeps API contracts stable even when tables change, and enables computed fields that don't exist in the database.

# models/user.py — ORM model (database representation)
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import declarative_base
import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, onupdate=datetime.datetime.utcnow)

# schemas/user.py — Pydantic schemas (API contract)
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional

class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8)
    full_name: str

class UserUpdate(BaseModel):
    full_name: Optional[str] = None
    password: Optional[str] = Field(default=None, min_length=8)

class UserResponse(BaseModel):
    id: int
    email: str
    full_name: Optional[str]
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}  # enables ORM → Pydantic conversion

Recommended Project Structure

After applying these patterns across dozens of production FastAPI projects, the following structure emerges as consistently effective. It balances clarity with pragmatism — deep enough to separate concerns, flat enough to avoid navigation fatigue.

project_root/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI app creation, middleware, startup events
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py           # Pydantic settings (BaseSettings)
│   │   ├── security.py         # JWT, password hashing, auth utilities
│   │   ├── exceptions.py       # Custom exception classes
│   │   └── dependencies.py     # Shared dependency providers
│   ├── models/                 # SQLAlchemy ORM models (one file per entity)
│   │   ├── __init__.py
│   │   ├── base.py             # Declarative base & mixins
│   │   ├── user.py
│   │   └── product.py
│   ├── schemas/                # Pydantic models (request/response shapes)
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── product.py
│   ├── repositories/           # Data access layer
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── user_repository.py
│   │   └── product_repository.py
│   ├── services/               # Business logic layer
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── product_service.py
│   ├── routers/                # API endpoints (one router per domain)
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── products.py
│   └── utils/                  # Shared helpers, pagination, response builders
│       ├── __init__.py
│       └── pagination.py
├── tests/
│   ├── unit/
│   │   ├── test_services/
│   │   └── test_repositories/
│   ├── integration/
│   └── conftest.py             # Test fixtures, async test helpers
├── alembic/                    # Database migrations
│   └── versions/
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
└── .env

How to Wire Everything Together

Below is a complete, runnable example that demonstrates how all layers connect in practice. Each file is shown in full so you can see exactly how imports flow and dependencies resolve.

Step 1: Application Entry Point

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.routers import users, products

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    openapi_url="/openapi.json"
)

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

app.include_router(users.router)
app.include_router(products.router)

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

Step 2: Configuration with Pydantic BaseSettings

# app/core/config.py
from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    PROJECT_NAME: str = "FastAPI Architecture Demo"
    VERSION: str = "1.0.0"
    DATABASE_URL: str = "postgresql+asyncpg://user:password@localhost:5432/db"
    ALLOWED_ORIGINS: List[str] = ["http://localhost:3000"]
    JWT_SECRET_KEY: str
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    model_config = {"env_file": ".env"}

settings = Settings()

Step 3: Async Database Session Setup

# app/core/dependencies.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.core.config import settings
from typing import AsyncGenerator

engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    echo=False
)

async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

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

Step 4: Custom Exceptions

# app/core/exceptions.py
class AppError(Exception):
    """Base application error."""
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class NotFoundError(AppError):
    def __init__(self, message: str = "Resource not found"):
        super().__init__(message, status_code=404)

class ConflictError(AppError):
    def __init__(self, message: str = "Resource already exists"):
        super().__init__(message, status_code=409)

class UnauthorizedError(AppError):
    def __init__(self, message: str = "Invalid credentials"):
        super().__init__(message, status_code=401)

Step 5: Exception Handlers in Main

# Add to app/main.py after app creation
from fastapi import Request
from fastapi.responses import JSONResponse
from app.core.exceptions import AppError

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message}
    )

Step 6: Base Repository with Generic Typing

# app/repositories/base.py
from typing import TypeVar, Generic, Optional, List, Sequence
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from sqlalchemy.sql.expression import ColumnElement

T = TypeVar('T')

class BaseRepository(Generic[T]):
    def __init__(self, session: AsyncSession, model: type[T]):
        self.session = session
        self.model = model

    async def get_by_id(self, id: int) -> Optional[T]:
        stmt = select(self.model).where(self.model.id == id)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def get_all(
        self, 
        skip: int = 0, 
        limit: int = 100,
        order_by: Optional[ColumnElement] = None
    ) -> List[T]:
        stmt = select(self.model)
        if order_by is not None:
            stmt = stmt.order_by(order_by)
        stmt = stmt.offset(skip).limit(limit)
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def count(self, *filters: ColumnElement) -> int:
        stmt = select(func.count()).select_from(self.model)
        if filters:
            stmt = stmt.where(*filters)
        result = await self.session.execute(stmt)
        return result.scalar_one()

    async def create(self, entity: T) -> T:
        self.session.add(entity)
        await self.session.flush()
        await self.session.refresh(entity)
        return entity

    async def update(self, entity: T) -> T:
        await self.session.flush()
        await self.session.refresh(entity)
        return entity

    async def delete(self, entity: T) -> None:
        await self.session.delete(entity)
        await self.session.flush()

Step 7: Domain-Specific Repository

# app/repositories/user_repository.py
from app.repositories.base import BaseRepository
from app.models.user import User
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Optional

class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)

    async def get_by_email(self, email: str) -> Optional[User]:
        stmt = select(User).where(User.email == email)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def get_active_users(self, skip: int = 0, limit: int = 100) -> list[User]:
        stmt = (
            select(User)
            .where(User.is_active == True)
            .offset(skip)
            .limit(limit)
        )
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

Step 8: Service Layer with Business Logic

# app/services/user_service.py
from app.repositories.user_repository import UserRepository
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate, UserResponse
from app.core.exceptions import NotFoundError, ConflictError
from app.core.security import hash_password, verify_password
from typing import List

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def create_user(self, data: UserCreate) -> UserResponse:
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError("A user with this email already exists")

        user = User(
            email=data.email,
            hashed_password=hash_password(data.password),
            full_name=data.full_name
        )
        created = await self.repo.create(user)
        return UserResponse.model_validate(created)

    async def get_user_by_id(self, user_id: int) -> UserResponse:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise NotFoundError(f"User with id {user_id} not found")
        return UserResponse.model_validate(user)

    async def list_users(self, skip: int = 0, limit: int = 100) -> List[UserResponse]:
        users = await self.repo.get_all(skip=skip, limit=limit)
        return [UserResponse.model_validate(u) for u in users]

    async def update_user(self, user_id: int, data: UserUpdate) -> UserResponse:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise NotFoundError(f"User with id {user_id} not found")

        update_dict = data.model_dump(exclude_unset=True)
        if "password" in update_dict:
            update_dict["hashed_password"] = hash_password(update_dict.pop("password"))

        for key, value in update_dict.items():
            if hasattr(user, key):
                setattr(user, key, value)

        updated = await self.repo.update(user)
        return UserResponse.model_validate(updated)

    async def delete_user(self, user_id: int) -> None:
        user = await self.repo.get_by_id(user_id)
        if not user:
            raise NotFoundError(f"User with id {user_id} not found")
        await self.repo.delete(user)

Step 9: Dependency Providers (Wire Everything)

# Add to app/core/dependencies.py or create app/dependencies/service_deps.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.dependencies import get_db
from app.repositories.user_repository import UserRepository
from app.services.user_service import UserService

async def get_user_repository(
    db: AsyncSession = Depends(get_db)
) -> UserRepository:
    return UserRepository(db)

async def get_user_service(
    repo: UserRepository = Depends(get_user_repository)
) -> UserService:
    return UserService(repo)

Step 10: Clean Router (Transport Layer Only)

# app/routers/users.py
from fastapi import APIRouter, Depends, Query, status
from app.services.user_service import UserService
from app.schemas.user import UserCreate, UserUpdate, UserResponse
from app.dependencies.service_deps import get_user_service
from typing import List

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

@router.post(
    "/",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create a new user"
)
async def create_user(
    data: UserCreate,
    service: UserService = Depends(get_user_service)
):
    return await service.create_user(data)

@router.get(
    "/",
    response_model=List[UserResponse],
    summary="List all users"
)
async def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=200),
    service: UserService = Depends(get_user_service)
):
    return await service.list_users(skip=skip, limit=limit)

@router.get(
    "/{user_id}",
    response_model=UserResponse,
    summary="Get a user by ID"
)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    return await service.get_user_by_id(user_id)

@router.patch(
    "/{user_id}",
    response_model=UserResponse,
    summary="Update a user"
)
async def update_user(
    user_id: int,
    data: UserUpdate,
    service: UserService = Depends(get_user_service)
):
    return await service.update_user(user_id, data)

@router.delete(
    "/{user_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete a user"
)
async def delete_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    await service.delete_user(user_id)

Best Practices Summary

• Keep routers thin — Route functions should contain zero business logic. They receive the request, call a service method, and return the response. If a route function exceeds 10 lines, extract logic into the service layer.
• Never expose ORM models in responses — Always use response_model with a dedicated Pydantic schema. Use model_config = {"from_attributes": True} to enable automatic ORM-to-Pydantic conversion without manual mapping.
• Use async database drivers consistently — Pair FastAPI's async event loop with asyncpg (PostgreSQL), aiomysql (MySQL), or aiosqlite (SQLite). Never mix sync and async database access in the same request path — it defeats the purpose of async and can cause thread pool exhaustion.
• Centralize exception handling — Define a custom exception hierarchy and register handlers at the app level. This keeps error responses consistent across all routes and eliminates repetitive try/except blocks.
• Use Pydantic BaseSettings for configuration — Load environment variables, secrets, and feature flags through a single settings object. Never hardcode values or scatter os.getenv calls throughout the codebase.
• Write tests at the service layer — Unit test services with mocked repositories for fast, deterministic tests. Reserve integration tests (hitting a real test database) for repository implementations and critical end-to-end flows.
• Apply middleware judiciously — CORS, request ID generation, and logging are appropriate as middleware. Business logic validation belongs in services, not middleware.
• Version your API from day one — Even if you only have v1, prefix routers with /api/v1/. This costs nothing upfront and saves painful migrations when breaking changes become necessary.
• Use dependency overrides for testing — FastAPI's app.dependency_overrides dictionary lets you swap real dependencies for test doubles without modifying any production code. This is cleaner than passing test flags through dependency chains.
• Keep modules focused and small — Each file should have a single clear responsibility. When a service file exceeds ~200 lines, consider splitting it into multiple services within a subpackage.

Testing with Dependency Overrides

One of FastAPI's most powerful testing features is the ability to override dependencies globally for a test session. Here's how to test the service layer with mocked repositories using this mechanism:

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.dependencies.service_deps import get_user_service
from unittest.mock import AsyncMock

@pytest.fixture
def mock_user_service():
    service = AsyncMock()
    service.create_user.return_value = {
        "id": 1,
        "email": "test@example.com",
        "full_name": "Test User",
        "is_active": True
    }
    return service

@pytest.fixture
def client(mock_user_service):
    app.dependency_overrides[get_user_service] = lambda: mock_user_service
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

# tests/test_users_api.py
def test_create_user_returns_201(client):
    response = client.post("/users", json={
        "email": "test@example.com",
        "password": "securepass123",
        "full_name": "Test User"
    })
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

Conclusion

The architecture patterns described here — router separation, dependency injection chains, repository abstraction, service layer isolation, and strict model-schema separation — form a cohesive blueprint for FastAPI applications that remain maintainable as they grow. The key insight is that FastAPI's design rewards inversion of control: route handlers don't create their own dependencies, they declare what they need and let the framework inject properly configured instances. This inversion, combined with disciplined separation between transport, business logic, and data access, yields a codebase where each layer can be tested in isolation, modified independently, and understood without tracing through the entire stack. Start with these patterns even on small projects, and you'll find that adding features, swapping infrastructure, and onboarding new team members remain smooth experiences rather than accumulating technical debt. The investment in structure pays for itself remarkably quickly in any application that lives beyond its initial prototype phase.