How I Structure a FastAPI Project for a Team of 5
The first FastAPI project I led for a team had everything in two files. main.py with the routes. models.py with the Pydantic schemas. It worked fine until a second engineer joined and we started stepping on each other constantly — merge conflicts on main.py every other day, no clear ownership of anything, and a growing pile of utility functions that lived wherever they'd been written first.
We restructured twice before landing on something that actually scaled with the team. Here's what we learned.
The Mistakes
Everything in main.py. Feels clean at the start. Becomes a 1,500-line file that nobody wants to open. Every feature touches it. Every PR conflicts with the one before it.
Models mixed with business logic. Pydantic schemas sitting next to database query functions sitting next to helper utilities. No clear layer boundaries. A junior engineer joining the team has no idea where to put a new thing.
No separation between internal and external schemas. Using the same Pydantic model for API input, database writes, and inter-service events. One schema change breaks three things at once.
Config scattered across files. Environment variables read directly in route handlers, service files, and utility modules. No single place to look at what the application needs to run.
The Structure That Fixed It
pulsecart-api/
├── main.py # app factory only — no routes, no logic
├── config.py # all env vars in one place
├── dependencies.py # shared FastAPI dependencies (db pool, auth, etc.)
│
├── routers/
│ ├── __init__.py
│ ├── events.py # POST /events/ingest
│ ├── orders.py # GET/POST /orders
│ └── health.py # GET /health, GET /ready
│
├── models/
│ ├── __init__.py
│ ├── base.py # PulseCartEvent base schema
│ ├── commerce.py # OrderPlaced, CartAbandoned, etc.
│ └── internal.py # schemas for inter-service communication
│
├── services/
│ ├── __init__.py
│ ├── publisher.py # Pub/Sub publishing logic
│ ├── idempotency.py # Redis deduplication
│ └── messaging.py # personalized message triggering
│
├── db/
│ ├── __init__.py
│ ├── pool.py # connection pool setup
│ └── queries/
│ ├── orders.py # SQL for order operations
│ └── events.py # SQL for event logging
│
└── tests/
├── conftest.py
├── unit/
└── integration/
Five folders, clear ownership. A new engineer can look at this and know exactly where to find something and where to put something new.
main.py — App Factory Only
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from routers import events, orders, health
from db.pool import init_pool, close_pool
from config import settings
import logging
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.db = await init_pool(settings.database_url)
yield
await close_pool(app.state.db)
app = FastAPI(
title="PulseCart API",
version="1.0.0",
lifespan=lifespan
)
app.include_router(events.router)
app.include_router(orders.router)
app.include_router(health.router)
main.py does three things: lifespan management, router registration, app config. Nothing else. If you're adding business logic here, it belongs in services/.
config.py — All Env Vars in One Place
# config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
redis_url: str
gcp_project_id: str
pubsub_topic_commerce: str = "pulsecart.commerce-events"
environment: str = "dev"
log_level: str = "INFO"
class Config:
env_file = ".env"
settings = Settings()
Every environment variable the application needs is declared here with a type and an optional default. If a required variable is missing, the app fails at startup with a clear error — not at runtime when the first request hits the code path that needs it.
No os.getenv() anywhere else in the codebase. If an engineer needs a new env var, it goes in config.py first.
dependencies.py — Shared FastAPI Dependencies
# dependencies.py
from fastapi import Request, HTTPException, status
from typing import AsyncGenerator
import asyncpg
async def get_db(request: Request) -> AsyncGenerator[asyncpg.Connection, None]:
async with request.app.state.db.acquire() as conn:
yield conn
async def get_current_service(request: Request) -> str:
api_key = request.headers.get("X-Service-Key")
if not api_key or api_key not in VALID_SERVICE_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing service key"
)
return api_key
Anything shared across routes — database connections, auth checks, request context — lives here and gets injected via FastAPI's dependency system. Routes stay clean. Tests swap these out via dependency_overrides.
routers/ — One File Per Domain
# routers/events.py
from fastapi import APIRouter, Depends, status
from models.commerce import EVENT_REGISTRY
from services.publisher import publish_event
from dependencies import get_db
import asyncpg
router = APIRouter(prefix="/events", tags=["events"])
@router.post("/ingest", status_code=status.HTTP_202_ACCEPTED)
async def ingest_event(
raw_event: dict,
db: asyncpg.Connection = Depends(get_db)
):
event_type = raw_event.get("event_type")
model_class = EVENT_REGISTRY.get(event_type)
event = model_class(**raw_event)
message_id = await publish_event(event.model_dump())
return {"status": "accepted", "event_id": event.event_id, "message_id": message_id}
Each router owns one domain. events.py owns event ingestion. orders.py owns order operations. When an engineer is working on orders, they touch routers/orders.py, services/ if needed, and db/queries/orders.py. They don't touch anything else.
services/ — Business Logic, No HTTP
Services contain the actual work. No Request objects, no Response objects, no FastAPI imports. Pure Python functions that take inputs and return outputs.
This matters for testing — services can be tested directly without spinning up an HTTP server, and without mocking the FastAPI layer.
# services/publisher.py
from google.cloud import pubsub_v1
from config import settings
import json
import logging
logger = logging.getLogger(__name__)
publisher = pubsub_v1.PublisherClient()
async def publish_event(event: dict) -> str:
topic_path = resolve_topic(event["event_type"])
future = publisher.publish(
topic_path,
data=json.dumps(event, default=str).encode(),
ordering_key=event["user_id"],
event_type=event["event_type"],
)
return future.result(timeout=10)
The Rules the Team Follows
Routes call services. Services call db queries. Services don't call routes. The dependency direction is one-way. A router can call a service. A service can call a db query. Nothing calls back up the chain.
No business logic in routes. A route handler should be readable in 10 lines. If it's longer, something belongs in services/.
No direct DB access outside db/. Every SQL query lives in db/queries/. Routes and services never construct SQL strings directly.
One model per concern. API input schema, database write schema, and event payload schema are separate models — even if they look similar today. They'll diverge, and when they do, having them separate costs nothing. Having them merged costs a refactor.
The structure above isn't the only valid one. It's the one that reduced merge conflicts, made onboarding faster, and gave every engineer on a five-person team a clear answer to "where does this go?"
That's the bar worth optimising for.
