Building Maintainable FastAPI Services

FastAPI makes it easy to build APIs quickly. But "quickly" often means "technical debt" if you don't establish good patterns early. This article covers the architecture patterns I use for FastAPI services that remain maintainable as they grow.

Project Structure

app/
├── main.py                 # App factory, middleware, lifespan
├── config.py               # Settings, environment
├── database.py             # DB connection, session management
├── exceptions.py           # Custom exceptions, handlers
├── dependencies.py         # Shared dependencies
├── models/                 # SQLAlchemy models
├── schemas/                # Pydantic schemas (request/response)
├── services/               # Business logic (single responsibility)
├── repositories/           # Data access layer
├── api/
│   ├── deps.py             # API-specific dependencies
│   ├── routes/
│   │   ├── users.py
│   │   ├── orders.py
│   │   └── ...
│   └── v1/
│       └── router.py
└── tests/
    ├── conftest.py
    ├── unit/
    ├── integration/
    └── fixtures/

Configuration Management

# config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # App
    app_name: str = "My Service"
    debug: bool = False
    api_prefix: str = "/api/v1"
    
    # Database
    database_url: str
    database_pool_size: int = 20
    database_max_overflow: int = 10
    
    # Redis
    redis_url: str
    
    # Security
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    
    # External services
    stripe_secret_key: str | None = None
    aws_access_key_id: str | None = None
    
    class Config:
        env_file = ".env"
        case_sensitive = False

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

settings = get_settings()

Database Layer

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from contextlib import asynccontextmanager

class Base(DeclarativeBase):
    pass

engine = create_async_engine(
    settings.database_url,
    pool_size=settings.database_pool_size,
    max_overflow=settings.database_max_overflow,
    pool_pre_ping=True,
    echo=settings.debug,
)

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

@asynccontextmanager
async def get_session() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# Dependency for FastAPI
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with get_session() as session:
        yield session

Repository Pattern

# repositories/base.py
from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

ModelType = TypeVar("ModelType")

class BaseRepository(Generic[ModelType], ABC):
    def __init__(self, session: AsyncSession, model: type[ModelType]):
        self.session = session
        self.model = model
    
    async def get(self, id: int) -> ModelType | None:
        return await self.session.get(self.model, id)
    
    async def get_all(self, *, skip: int = 0, limit: int = 100) -> Sequence[ModelType]:
        result = await self.session.execute(
            select(self.model).offset(skip).limit(limit)
        )
        return result.scalars().all()
    
    async def create(self, **kwargs) -> ModelType:
        obj = self.model(**kwargs)
        self.session.add(obj)
        await self.session.flush()
        return obj
    
    async def update(self, id: int, **kwargs) -> ModelType | None:
        obj = await self.get(id)
        if not obj:
            return None
        for key, value in kwargs.items():
            setattr(obj, key, value)
        await self.session.flush()
        return obj
    
    async def delete(self, id: int) -> bool:
        obj = await self.get(id)
        if not obj:
            return False
        await self.session.delete(obj)
        return True

# repositories/user.py
from models.user import User
from schemas.user import UserCreate, UserUpdate

class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)
    
    async def get_by_email(self, email: str) -> User | None:
        result = await self.session.execute(
            select(User).where(User.email == email)
        )
        return result.scalar_one_or_none()
    
    async def get_by_ids(self, ids: list[int]) -> Sequence[User]:
        result = await self.session.execute(
            select(User).where(User.id.in_(ids))
        )
        return result.scalars().all()
    
    async def search(self, query: str, *, skip: int = 0, limit: int = 20) -> Sequence[User]:
        result = await self.session.execute(
            select(User)
            .where(User.email.ilike(f"%{query}%") | User.name.ilike(f"%{query}%"))
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all()

Service Layer

# services/user.py
from repositories.user import UserRepository
from schemas.user import UserCreate, UserUpdate, UserResponse
from exceptions import NotFoundError, ConflictError
from security import hash_password, verify_password

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo
    
    async def create_user(self, data: UserCreate) -> UserResponse:
        # Check uniqueness
        if await self.repo.get_by_email(data.email):
            raise ConflictError("Email already registered")
        
        # Hash password
        hashed = hash_password(data.password)
        
        # Create user
        user = await self.repo.create(
            email=data.email,
            name=data.name,
            hashed_password=hashed,
        )
        return UserResponse.model_validate(user)
    
    async def authenticate(self, email: str, password: str) -> UserResponse | None:
        user = await self.repo.get_by_email(email)
        if not user or not verify_password(password, user.hashed_password):
            return None
        return UserResponse.model_validate(user)
    
    async def get_user(self, user_id: int) -> UserResponse:
        user = await self.repo.get(user_id)
        if not user:
            raise NotFoundError("User not found")
        return UserResponse.model_validate(user)
    
    async def update_user(self, user_id: int, data: UserUpdate) -> UserResponse:
        user = await self.repo.get(user_id)
        if not user:
            raise NotFoundError("User not found")
        
        # Check email uniqueness if changing
        if data.email and data.email != user.email:
            if await self.repo.get_by_email(data.email):
                raise ConflictError("Email already in use")
        
        update_data = data.model_dump(exclude_unset=True)
        user = await self.repo.update(user_id, **update_data)
        return UserResponse.model_validate(user)

Dependency Injection

# dependencies.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from repositories.user import UserRepository
from services.user import UserService

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

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

# Reusable auth dependency
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    service: UserService = Depends(get_user_service)
) -> UserResponse:
    credentials_exception = HTTPException(
        status_code=401, detail="Could not validate credentials"
    )
    payload = decode_token(token)
    user = await service.get_user(payload.sub)
    if not user:
        raise credentials_exception
    return user

API Routes

# api/routes/users.py
from fastapi import APIRouter, Depends, status
from schemas.user import UserCreate, UserUpdate, UserResponse, UserList
from services.user import UserService
from dependencies import get_user_service, get_current_user

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

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

@router.get("/me", response_model=UserResponse)
async def get_me(current_user: UserResponse = Depends(get_current_user)):
    return current_user

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    return await service.get_user(user_id)

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

@router.get("", response_model=UserList)
async def list_users(
    skip: int = 0,
    limit: int = 20,
    service: UserService = Depends(get_user_service)
):
    users = await service.list_users(skip=skip, limit=limit)
    total = await service.count_users()
    return UserList(users=users, total=total, skip=skip, limit=limit)

Error Handling

# exceptions.py
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from pydantic import ValidationError

class AppException(Exception):
    def __init__(self, message: str, status_code: int = 500):
        self.message = message
        self.status_code = status_code

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

class ConflictError(AppException):
    def __init__(self, message: str = "Resource conflict"):
        super().__init__(message, 409)

# Exception handlers
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message, "type": exc.__class__.__name__}
    )

async def validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "type": "ValidationError"}
    )

# Register in main.py
app.add_exception_handler(AppException, app_exception_handler)
app.add_exception_handler(ValidationError, validation_exception_handler)

Testing Strategy

# tests/conftest.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.pool import StaticPool
from app.main import app
from app.database import get_db, Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def engine():
    engine = create_async_engine(
        TEST_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    return engine

@pytest.fixture(scope="function")
async def db_session(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    async_session = async_sessionmaker(engine, expire_on_commit=False)
    async with async_session() as session:
        yield session
    
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session
    
    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client
    app.dependency_overrides.clear()

# tests/integration/test_users.py
async def test_create_user(client):
    response = await client.post("/api/v1/users", json={
        "email": "test@example.com",
        "name": "Test User",
        "password": "securepassword123"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
    assert "id" in data

async def test_get_user_not_found(client):
    response = await client.get("/api/v1/users/999")
    assert response.status_code == 404
    assert response.json()["type"] == "NotFoundError"

Key Principles Summary

| Layer | Responsibility | Testing | |-------|---------------|---------| | Routes | HTTP concerns, serialization, validation | Integration tests | | Services | Business logic, orchestration | Unit tests (mock repo) | | Repositories | Data access, queries | Integration tests (real DB) | | Models | Schema definition, relationships | N/A | | Schemas | Validation, serialization | Unit tests |

This architecture scales because:

• Each layer has one reason to change
• Dependencies point inward (routes → services → repos → models)
• Easy to test at every level
• Explicit contracts via Pydantic schemas
• Framework-agnostic business logic

---

Next: How AI Is Changing the Way We Build Software — exploring AI-assisted development workflows.