🎯 What You Will Learn
How to grow a FastAPI app past the one-giant-main.py stage into a clean, layered structure that stays easy to read, test, and extend — the same layering used in production codebases.
- The router → service → repository layering
- Type-based vs feature-based folder layout (and which to pick)
- Splitting features with
APIRouter - Where business logic, data access, and HTTP each belong
- Wiring layers together with dependencies
- How to add a new feature without touching the old ones
Prerequisites: comfort with Depends() (fastapi-dependency-injection) and a database session (fastapi-async-sqlalchemy or fastapi-sqlmodel). We reuse both ideas here.
The Layers: Router → Service → Repository
The core idea is separation of concerns: each layer has one job and talks only to the layer below it. HTTP concerns stay at the top; database concerns stay at the bottom; business rules sit in the middle.
HTTP request
|
v
[ Router ] -> HTTP only: parse input, call service, return response
|
v
[ Service ] -> business logic: rules, validation, orchestration
|
v
[ Repository ] -> data access: the ONLY layer that talks to the DB
|
v
Database| Layer | Knows about | Never touches |
|---|---|---|
| Router | HTTP, status codes, request/response schemas | SQL / ORM queries |
| Service | Business rules, domain exceptions | Request objects, status codes |
| Repository | The database / ORM | HTTP, business rules |
💡
Ang importante dito: if you can swap Postgres for another store by editing only the repository, or reuse the service from a CLI with no HTTP, your layering is right.
The Problem: Everything in main.py (See It First)
Every project starts here. It's fine for a demo — and painful the moment it grows:
app/main.py (everything mixed together)
@app.post("/users")
def create_user(payload: dict, session=Depends(get_session)):
# HTTP parsing, business rules, AND raw DB access — all in one place
if session.query(User).filter_by(email=payload["email"]).first():
raise HTTPException(status_code=409, detail="Email taken")
user = User(username=payload["username"], email=payload["email"])
session.add(user)
session.commit()
return user
# ...and 40 more endpoints just like it in the same file 😩⚠️
Why this hurts: one 2,000-line file, merge conflicts on every PR, business rules you can't test without spinning up HTTP, and DB queries copy-pasted between endpoints. Structure is what stops this.
Type-Based vs Feature-Based Layout
There are two common ways to organize the folders. Pick based on how the app will grow.
| Layout | Groups by | Best for |
|---|---|---|
| Type-based | routers/, services/, models/ (all routers together, etc.) | Small apps, tutorials, few features |
| Feature-based | users/, orders/, auth/ (each holds its own router+service+repo) | Growing apps — everything for a feature lives together |
💡
Recommendation: use feature-based for anything that will grow. Adding a feature means adding a folder, not editing five shared folders — and you can find everything about "users" in one place.
The Folder Layout
Here's the feature-based layout we'll build. Shared plumbing lives in core/; each feature is self-contained.
app/
├── main.py # create app, include the API router
├── api.py # aggregates every feature router
├── core/
│ ├── config.py # settings (pydantic-settings)
│ └── database.py # engine + get_session dependency
└── users/ # a self-contained feature
├── __init__.py
├── models.py # ORM model
├── schemas.py # Pydantic request/response
├── repository.py # data access
├── service.py # business logic
├── router.py # HTTP endpoints
└── dependencies.py# wiring: session -> repo -> service💡
Adding an
orders feature? Copy the shape: a new app/orders/ folder with its own router, service, and repository. Nothing else changes.The Repository Layer (Data Access)
The repository is the only place that talks to the database. It exposes intent-revealing methods (get_by_email), not raw queries scattered across the app.
app/users/models.py
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(unique=True)
email: Mapped[str] = mapped_column(unique=True, index=True)app/users/repository.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.users.models import User
class UserRepository:
def __init__(self, session: AsyncSession) -> None:
self.session = session
async def get_by_id(self, user_id: int) -> User | None:
return await self.session.get(User, user_id)
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 add(self, user: User) -> User:
self.session.add(user)
await self.session.commit()
await self.session.refresh(user)
return user💡
Why a class? It holds the session and groups all user queries. Swap the storage engine or mock it in a test by replacing this one class — nothing above it changes.
The Service Layer (Business Logic)
The service enforces the rules: "emails must be unique." It raises domain exceptions (see fastapi-error-handling) and knows nothing about HTTP or status codes.
app/users/service.py
from app.users.models import User
from app.users.repository import UserRepository
from app.users.schemas import UserCreate
class EmailAlreadyExists(Exception):
"""Raised when creating a user with a taken email."""
class UserNotFound(Exception):
"""Raised when a user id doesn't exist."""
class UserService:
def __init__(self, repository: UserRepository) -> None:
self.repository = repository
async def register(self, data: UserCreate) -> User:
if await self.repository.get_by_email(data.email):
raise EmailAlreadyExists(data.email)
user = User(username=data.username, email=data.email)
return await self.repository.add(user)
async def get(self, user_id: int) -> User:
user = await self.repository.get_by_id(user_id)
if user is None:
raise UserNotFound(str(user_id))
return user✅
Framework-free logic:
UserService has zero FastAPI imports. You can call register() from a script, a Celery task, or a unit test — no HTTP required.Wiring the Layers with Dependencies
Dependencies build the chain: a session creates a repository, which creates a service. Each route just asks for the service.
app/users/dependencies.py
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_session
from app.users.repository import UserRepository
from app.users.service import UserService
def get_user_service(
session: Annotated[AsyncSession, Depends(get_session)],
) -> UserService:
repository = UserRepository(session)
return UserService(repository)
UserServiceDep = Annotated[UserService, Depends(get_user_service)]The Router Layer (HTTP)
The router is thin: it maps HTTP to service calls and translates domain exceptions into status codes. No business rules, no SQL.
app/users/schemas.py
from pydantic import BaseModel, ConfigDict, EmailStr
class UserCreate(BaseModel):
username: str
email: EmailStr
class UserRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
username: str
email: EmailStrapp/users/router.py
from fastapi import APIRouter, HTTPException, status
from app.users.dependencies import UserServiceDep
from app.users.schemas import UserCreate, UserRead
from app.users.service import EmailAlreadyExists, UserNotFound
router = APIRouter(prefix="/users", tags=["users"])
@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(data: UserCreate, service: UserServiceDep):
try:
return await service.register(data)
except EmailAlreadyExists:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already exists")
@router.get("/{user_id}", response_model=UserRead)
async def get_user(user_id: int, service: UserServiceDep):
try:
return await service.get(user_id)
except UserNotFound:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")💡
Even cleaner: register global exception handlers (from
fastapi-error-handling) for EmailAlreadyExists / UserNotFound and the router loses even the try/except— the mapping lives in one place.Aggregating Routers + main.py
One api.py collects every feature router behind a shared prefix like /api/v1. main.py stays tiny — it just creates the app and includes that one router.
app/api.py
from fastapi import APIRouter
from app.users.router import router as users_router
# from app.orders.router import router as orders_router # future feature
api_router = APIRouter(prefix="/api/v1")
api_router.include_router(users_router)
# api_router.include_router(orders_router)app/main.py
from fastapi import FastAPI
from app.api import api_router
app = FastAPI(title="Scalable FastAPI", version="0.1.0")
app.include_router(api_router)
@app.get("/health")
def health():
return {"status": "ok"}Routes now live under the versioned prefix:
POST /api/v1/users
GET /api/v1/users/{user_id}
GET /health✅
Payoff:
main.py never grows. Every new feature adds a folder and one include_router line — the rest of the app is untouched.Adding a New Feature (The Whole Point)
Want an orders feature? The recipe is mechanical:
- Create
app/orders/with the same six files. - Write
OrderRepository(queries),OrderService(rules),router.py(endpoints). - Add one line to
app/api.py:api_router.include_router(orders_router).
💡
No existing file's logic changes — you only append one router include. That's the difference between a codebase that scales and one that fights you.
Common Mistakes (and Fixes)
| Mistake | Symptom | Fix |
|---|---|---|
| SQL queries inside routers | DB logic duplicated across endpoints | Move all queries into the repository |
| Business rules in the router | Can't test logic without HTTP | Put rules in the service; keep routers thin |
Service raising HTTPException | Business layer coupled to the web framework | Raise domain exceptions; map to HTTP in the router |
| Over-engineering a tiny app | Layers for 3 endpoints slow you down | Start type-based; refactor to feature-based as it grows |
| Circular imports between layers | ImportError on startup | Depend downward only (router→service→repo) |
Giant main.py | Merge conflicts, hard to navigate | Aggregate routers via api.py; keep main.py tiny |
⚠️
Don't cargo-cult it: a 3-endpoint service doesn't need five layers. Structure should match complexity. The signal to add layers is real pain — duplication, big files, untestable logic — not a blog post telling you to.
What's Next
✅
🚀 Recommended next reads:
fastapi-error-handling— global handlers so routers drop thetry/exceptfastapi-dependency-injection— the wiring pattern behind the service dependencyfastapi-async-sqlalchemy— the session the repository depends onfastapi-integration-testing— test services directly, no HTTP needed
Recap: keep HTTP in the router, rules in the service, and data access in the repository. Group by feature so each folder is self-contained, aggregate routers in one place, and keep main.py tiny. Add features by adding folders — never by growing one file.