Backend · Architecture

The FastAPI service layout I use on every new project

Abstract cover: three stacked layers with a request line passing through, in the site's dark palette

I have started five FastAPI codebases in production anger: the therapy-session and Riri backends at Relaxy, service layers at BDChat, and the AI products we are building now at Mangosteen. Every one of them begins from the same skeleton. The two that didn't — inherited codebases where "everything is in main.py" — both got migrated to it within a quarter, not because the layout is clever, but because every alternative eventually cost us a production incident.

FastAPI itself won't push you toward any structure. The tutorial puts the route, the validation, the query and the business rule in one function, and it works beautifully right up until the second engineer joins. This post is the structure that survived the second engineer, and the third team.

The skeleton

app/
  api/
    routes/        # routers only — no business rules
      chat.py
      billing.py
    deps.py        # Depends() providers, auth, pagination
  services/        # decisions live here
    chat.py
  repositories/    # SQL lives here, and nowhere else
    messages.py
  schemas/         # pydantic request/response models
  core/
    settings.py    # the only file that reads the environment
    errors.py      # one error envelope for the whole API
  main.py          # app factory, middleware, router mounting

Three layers, one direction. A request enters a router, the router calls a service, the service calls repositories. Nothing ever calls upward, and nothing skips a layer on the write path.

A request flowing through router, service and repository layers into PostgresRouterServiceRepositoryPostgresvalidates shapemakes decisionsowns the SQLstores facts
One direction of flow. The service layer is the only place allowed to say no.

Step 1 — routers validate shape, nothing else

A router's whole job is to turn HTTP into a typed function call and back. If you can read a router and learn anything about the business, something has leaked.

# api/routes/chat.py
router = APIRouter(prefix="/chat", tags=["chat"])

@router.post("/messages", response_model=MessageOut, status_code=201)
async def send_message(
    body: MessageIn,
    user: CurrentUser,
    chat: ChatService = Depends(get_chat_service),
) -> MessageOut:
    return await chat.send(user_id=user.id, text=body.text)

One line of body. Pydantic already rejected malformed input before this function ran; auth already happened in CurrentUser. The test for "is a blocked user allowed to message" does not live here, so it will never need an HTTP client to run.

Step 2 — services make every decision

The service layer is plain Python classes with their dependencies passed into the constructor. No FastAPI imports, no request objects, no SQL.

# services/chat.py
class ChatService:
    def __init__(self, messages: MessageRepo, moderation: ModerationClient):
        self._messages = messages
        self._moderation = moderation

    async def send(self, *, user_id: UUID, text: str) -> MessageOut:
        verdict = await self._moderation.check(text)
        if verdict.blocked:
            raise DomainError("message_blocked", detail=verdict.reason)
        row = await self._messages.insert(user_id=user_id, text=text)
        return MessageOut.model_validate(row)

At Relaxy this boundary was not academic. Riri's send path had moderation, crisis-keyword detection and rate rules stacked in front of the model call. Because all of it lived in one service class, we could test the entire safety chain with stubs in milliseconds — and we ran those tests on every commit, because in a mental-health product that chain is the product.

If a rule matters enough to page you when it breaks, it matters enough to test without a web server.

Step 3 — repositories own the SQL

Every query lives in a repository, and repositories return rows or domain objects — never leak the ORM session upward. The payoff is boring and constant: when a query gets slow, there is exactly one file to open, and when we swapped a table's schema, the diff touched one layer.

# repositories/messages.py
class MessageRepo:
    def __init__(self, db: AsyncSession):
        self._db = db

    async def insert(self, *, user_id: UUID, text: str) -> MessageRow:
        stmt = (
            insert(messages)
            .values(user_id=user_id, text=text)
            .returning(messages)
        )
        result = await self._db.execute(stmt)
        return MessageRow(**result.mappings().one())

One settings module, frozen

Configuration bugs are deployment bugs, and deployment bugs happen at the worst hour. So exactly one file reads the environment, validates it at import time, and everything else asks for the typed object.

# core/settings.py
class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", frozen=True)

    database_url: PostgresDsn
    openai_api_key: SecretStr
    sentry_dsn: str | None = None

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

If a required variable is missing, the pod fails to boot with a named error instead of failing to charge a customer four hours later. frozen=True because a setting that can be mutated at runtime is a global variable with better marketing.

One error envelope

Every domain failure is a DomainError with a machine-readable code, and one handler turns them all into the same JSON shape. Mobile clients — and I say this having sat on both sides — do not want your stack trace; they want a stable code they can switch on.

# core/errors.py
@app.exception_handler(DomainError)
async def domain_error_handler(_: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status,
        content={"error": {"code": exc.code, "detail": exc.detail}},
    )

The test that proves the layout works

The whole point of the boundaries is this test. No HTTP, no database, no fixtures folder — and it covers the rule that actually matters:

async def test_blocked_message_never_reaches_the_db():
    svc = ChatService(
        messages=InMemoryMessageRepo(),
        moderation=StubModeration(blocked=True),
    )

    with pytest.raises(DomainError) as err:
        await svc.send(user_id=uuid4(), text="anything")

    assert err.value.code == "message_blocked"
    assert svc._messages.rows == []

Our service-layer suite at Relaxy ran in under four seconds. Engineers ran it reflexively, which is the only test suite metric that matters.

Where I break my own rules

Two places, both on purpose.

The layout is not the achievement. The achievement is that on three different teams, a new engineer's first pull request landed in the right file without anyone telling them where the right file was. Structure you have to explain is structure that isn't working.

All postsReply by email

Keep reading

Next posts