# 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

```plaintext
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

```python
# 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

```python
# 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

```python
# 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

```python
# 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.

```python
# 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.