Production FastAPI Project Structure at Scale
As a FastAPI project grows, a flat file structure becomes unmaintainable. Models, routes, business logic, and database code mix together. Adding a feature requires touching multiple files; refactoring breaks things invisibly. A layered architecture separates concerns: routes (API contracts), services (business logic), models (data), and repositories (database access). This structure lets teams work independently, test components in isolation, and scale from a prototype to a production system serving millions of requests.
I've scaled FastAPI projects from single files to monoliths with 100+ routes across dozens of files. This guide shows you a structure that works at any scale.
The Layered Architecture
A production FastAPI project has clear layers:
app/
__init__.py
main.py # Entry point; creates app
config.py # Settings & configuration
api/ # Routes (API contracts)
__init__.py
routers/
__init__.py
users.py # User endpoints
products.py # Product endpoints
orders.py # Order endpoints
services/ # Business logic
__init__.py
user_service.py # User operations
product_service.py # Product operations
order_service.py # Order operations
models/ # Data models (Pydantic + SQLAlchemy)
__init__.py
user.py # User model
product.py # Product model
order.py # Order model
repositories/ # Database access (DAOs)
__init__.py
base.py # Base repository class
user_repo.py # User queries
product_repo.py # Product queries
order_repo.py # Order queries
schemas/ # Pydantic request/response schemas
__init__.py
user_schemas.py
product_schemas.py
order_schemas.py
middleware/ # Custom middleware
__init__.py
logging.py # Logging middleware
auth.py # Authentication middleware
dependencies/ # Dependency injection
__init__.py
database.py # Database session
auth.py # Authentication
utils/ # Utilities (no business logic)
__init__.py
validators.py
formatters.py
Data flows: Request → Router → Service → Repository → Database. Each layer has a single responsibility.
Layer Definitions
Routes (API layer): Define HTTP endpoints, parse requests, call services.
# api/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from services.user_service import UserService
from schemas.user_schemas import UserCreate, UserResponse
from dependencies.database import get_session
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/", response_model=list[UserResponse])
async def list_users(skip: int = 0, limit: int = 10, db = Depends(get_session)):
"""List all users."""
service = UserService(db)
return await service.list_users(skip, limit)
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate, db = Depends(get_session)):
"""Create a new user."""
service = UserService(db)
return await service.create_user(user)
Routes are thin—they handle HTTP concerns (status codes, response schemas) and delegate logic to services.
Services (business logic layer): Implement business rules, orchestrate repositories, handle transactions.
# services/user_service.py
from sqlalchemy.ext.asyncio import AsyncSession
from repositories.user_repo import UserRepository
from schemas.user_schemas import UserCreate, UserResponse
from fastapi import HTTPException, status
class UserService:
def __init__(self, db: AsyncSession):
self.db = db
self.repo = UserRepository(db)
async def list_users(self, skip: int = 0, limit: int = 10):
"""Fetch users with pagination."""
return await self.repo.get_all(skip=skip, limit=limit)
async def create_user(self, user: UserCreate) -> UserResponse:
"""Create a new user with validation."""
# Business logic: check email uniqueness
existing = await self.repo.get_by_email(user.email)
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="User with this email already exists"
)
# Hash password
user.password = hash_password(user.password)
# Persist
db_user = await self.repo.create(user)
# Post-create logic (e.g., send welcome email)
await send_welcome_email(db_user.email)
return db_user
async def get_user(self, user_id: int) -> UserResponse:
"""Fetch a user, raising 404 if not found."""
user = await self.repo.get_by_id(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
Services contain business logic; they're independent of FastAPI and easily testable.
Repositories (data access layer): Query the database, abstract storage details.
# repositories/user_repo.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models.user import User
from schemas.user_schemas import UserCreate
class UserRepository:
def __init__(self, db: AsyncSession):
self.db = db
async def get_all(self, skip: int = 0, limit: int = 10):
"""Fetch all users with pagination."""
query = select(User).offset(skip).limit(limit)
result = await self.db.execute(query)
return result.scalars().all()
async def get_by_id(self, user_id: int):
"""Fetch a user by ID."""
return await self.db.get(User, user_id)
async def get_by_email(self, email: str):
"""Fetch a user by email."""
query = select(User).where(User.email == email)
result = await self.db.execute(query)
return result.scalars().first()
async def create(self, user: UserCreate) -> User:
"""Insert a new user."""
db_user = User(**user.dict())
self.db.add(db_user)
await self.db.commit()
await self.db.refresh(db_user)
return db_user
async def update(self, user_id: int, user: UserCreate) -> User:
"""Update an existing user."""
db_user = await self.get_by_id(user_id)
if not db_user:
return None
for key, value in user.dict().items():
setattr(db_user, key, value)
await self.db.commit()
await self.db.refresh(db_user)
return db_user
async def delete(self, user_id: int):
"""Delete a user."""
db_user = await self.get_by_id(user_id)
if db_user:
await self.db.delete(db_user)
await self.db.commit()
return db_user
Repositories are queries; they're testable with mocks.
Models and Schemas: SQLAlchemy ORM models (database representation) vs. Pydantic schemas (API contracts).
# models/user.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String, unique=True)
name = Column(String)
password_hash = Column(String)
# schemas/user_schemas.py
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
email: EmailStr
name: str
password: str
class UserResponse(BaseModel):
id: int
email: str
name: str
class Config:
from_attributes = True # Read from ORM models
Separate schemas from models to evolve your API independently of the database.
Dependency Injection and Initialization
Centralize dependency creation:
# dependencies/database.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from config import settings
SessionLocal = async_sessionmaker(...)
async def get_session() -> AsyncSession:
"""Dependency: provide database session."""
async with SessionLocal() as session:
yield session
# dependencies/auth.py
from fastapi.security import HTTPBearer
from dependencies.database import get_session
async def get_current_user(
credentials: HTTPAuthCredentials = Depends(HTTPBearer()),
db: AsyncSession = Depends(get_session)
):
"""Dependency: authenticate and resolve user."""
# Verify token, fetch user
return user
Wire dependencies in routes:
@router.get("/me")
async def get_current_user_profile(
user = Depends(get_current_user),
db = Depends(get_session)
):
service = UserService(db)
return await service.get_user(user.id)
Main Application Entry Point
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import settings
from api.routers import users, products, orders
from middleware.logging import LoggingMiddleware
from middleware.auth import AuthMiddleware
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
logger.info("Starting application")
# Initialize resources (pools, etc.)
yield
# Shutdown
logger.info("Shutting down")
# Cleanup
app = FastAPI(
title=settings.api_title,
version=settings.api_version,
lifespan=lifespan
)
# Add middleware
app.add_middleware(LoggingMiddleware)
app.add_middleware(AuthMiddleware)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"]
)
# Include routers
app.include_router(users.router)
app.include_router(products.router)
app.include_router(orders.router)
@app.get("/health")
async def health():
"""Liveness check."""
return {"status": "ok"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Testing with Layered Architecture
Each layer is independently testable:
# tests/unit/test_user_service.py
import pytest
from services.user_service import UserService
from repositories.user_repo import UserRepository
from unittest.mock import AsyncMock
@pytest.fixture
def mock_repo():
return AsyncMock(spec=UserRepository)
@pytest.mark.asyncio
async def test_create_user_success(mock_repo):
"""Test service creates user and sends email."""
service = UserService(db=None)
service.repo = mock_repo
user_data = UserCreate(email="[email protected]", name="Jane", password="pwd123")
mock_repo.get_by_email.return_value = None
mock_repo.create.return_value = User(id=1, email="[email protected]", name="Jane")
result = await service.create_user(user_data)
assert result.id == 1
mock_repo.create.assert_called_once()
# tests/integration/test_user_routes.py
from fastapi.testclient import TestClient
def test_list_users_api(client: TestClient):
"""Test list users endpoint."""
response = client.get("/users")
assert response.status_code == 200
assert isinstance(response.json(), list)
Configuration for Scale
As you scale, externalize configuration:
# config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# Database
database_url: str
database_pool_size: int = 20
# Security
secret_key: str
api_version: str = "1.0.0"
# Features (enable/disable features per environment)
enable_cache: bool = False
enable_analytics: bool = True
class Config:
env_file = ".env"
settings = Settings()
Use feature flags for gradual rollouts:
@router.get("/new-feature")
async def new_feature(db = Depends(get_session)):
if not settings.enable_new_feature:
raise HTTPException(status_code=404)
# ...
Key Takeaways
- Organize code in layers: routes, services, repositories, models.
- Routes handle HTTP; services handle business logic; repositories handle queries.
- Separate Pydantic schemas from SQLAlchemy models for API flexibility.
- Use dependency injection to wire components; easy to test and mock.
- Centralize configuration in a config module; use feature flags for gradual rollouts.
- Test each layer independently; integration tests verify layer interactions.
Frequently Asked Questions
Should every route have a service?
Yes. Services are cheap—they encapsulate logic and are easily testable. Even simple CRUD operations benefit from a service layer.
How do I handle cross-cutting concerns (auth, logging)?
Use middleware for global concerns (logging, timing). Use dependencies for route-specific concerns (authentication).
When should I split into microservices?
When a single service becomes too large (> 5,000 lines), has independent deployments, or separate scaling needs. Start monolithic; split when justified.
How do I test repositories without a real database?
Use unittest.mock.AsyncMock to mock the SQLAlchemy session. Or create a separate test database (in-memory SQLite) for integration tests.
Can I reuse services across routes?
Yes. If two routes share logic, extract it to a shared service. For example, both GET /users/{user_id} and GET /profile (current user) might call UserService.get_user().