Layered Architecture
As your FastAPI application grows, putting all logic in route handlers leads to unmaintainable code. A layered architecture separates concerns: routers handle HTTP, services contain business logic, repositories handle data access.
Learning Focus
By the end of this lesson you can: design a scalable FastAPI project directory tree, describe each layer's responsibility, and refactor spaghetti routes into clean layered code.
The Layers
| Layer | Responsibility | Should not |
|---|---|---|
| Router | Parse request, call service, return response | Contain business logic |
| Service | Orchestrate business rules, call repos | Know about HTTP or SQLAlchemy internals |
| Repository | Read/write database only | Apply business rules |
| Model | Define data shapes | Contain logic beyond validation |
Full Project Directory Tree
app/
├── main.py # FastAPI app instance, lifespan, router registration
├── core/
│ ├── config.py # Pydantic Settings class
│ ├── security.py # JWT, password hashing helpers
│ └── exceptions.py # Custom exception classes
├── routers/
│ ├── __init__.py
│ ├── auth.py # POST /auth/token, POST /auth/register
│ ├── users.py # /users CRUD
│ ├── products.py # /products CRUD
│ └── orders.py # /orders CRUD + status transitions
├── services/
│ ├── __init__.py
│ ├── user_service.py # User registration, profile, deactivation
│ ├── product_service.py # Catalog, inventory, pricing logic
│ └── order_service.py # Order placement, payment, fulfillment
├── repositories/
│ ├── __init__.py
│ ├── base.py # Generic CRUDBase class
│ ├── user_repo.py # User-specific queries
│ ├── product_repo.py # Product queries, search
│ └── order_repo.py # Order queries, status filter
├── models/
│ ├── __init__.py
│ ├── user.py # UserCreate, UserUpdate, UserResponse
│ ├── product.py # ProductCreate, ProductUpdate, ProductResponse
│ └── order.py # OrderCreate, OrderResponse, OrderStatus
├── db/
│ ├── __init__.py
│ ├── base.py # DeclarativeBase
│ ├── session.py # Engine, session factory, get_db
│ └── orm/
│ ├── user.py # SQLAlchemy User ORM model
│ ├── product.py # SQLAlchemy Product ORM model
│ └── order.py # SQLAlchemy Order ORM model
├── dependencies/
│ ├── __init__.py
│ ├── auth.py # get_current_user, require_role
│ ├── pagination.py # Pagination, sorting deps
│ └── rate_limit.py # Redis rate limiter dep
├── middleware/
│ ├── request_id.py # Request ID injection
│ └── security_headers.py # Security header middleware
└── cache/
└── redis_cache.py # get_cached, set_cached, invalidate
Service Layer Example
app/services/user_service.py
from app.repositories.user_repo import UserRepository
from app.models.user import UserCreate, UserResponse
from app.core.security import hash_password
from app.core.exceptions import DuplicateEmailError
class UserService:
def __init__(self, repo: UserRepository) -> None:
self._repo = repo
async def register(self, data: UserCreate) -> UserResponse:
existing = await self._repo.get_by_email(data.email)
if existing:
raise DuplicateEmailError(data.email)
hashed = hash_password(data.password)
user = await self._repo.create(
username=data.username,
email=data.email,
hashed_password=hashed,
)
return UserResponse.model_validate(user)
async def deactivate(self, user_id: int) -> None:
user = await self._repo.get_by_id(user_id)
if not user:
raise UserNotFoundError(user_id)
await self._repo.update(user_id, {"is_active": False})
Router Calling Service
app/routers/users.py
from typing import Annotated
from fastapi import APIRouter, Depends
from app.services.user_service import UserService
from app.dependencies.auth import CurrentUser
router = APIRouter(prefix="/users", tags=["users"])
def get_user_service(db: DBSession) -> UserService:
from app.repositories.user_repo import UserRepository
return UserService(UserRepository(db))
UserSvc = Annotated[UserService, Depends(get_user_service)]
@router.post("/", response_model=UserResponse, status_code=201)
async def register(data: UserCreate, svc: UserSvc) -> UserResponse:
return await svc.register(data)
Common Pitfalls
| Pitfall | Cause / Symptom | Fix |
|---|---|---|
| Business logic in router | Hard to test without HTTP | Move logic to service layer |
| SQLAlchemy ORM in service | Service tightly coupled to DB | Service calls repository methods only |
| Circular imports | Models import services, services import models | Use TYPE_CHECKING and string annotations |
| Too many layers for a small app | Excessive boilerplate | For small APIs, router + CRUD is fine; add layers as complexity grows |
| Repository returns ORM objects to router | Router coupled to DB layer | Repository returns Pydantic models or dicts |