The Repository Pattern in a FastAPI App: Decoupling Routes From SQLAlchemy

Scattering session.query(...) and select(...) calls through your route handlers welds your API to SQLAlchemy. Every handler becomes a place that knows about tables, joins and session lifecycle — and every test has to spin up a database to exercise a line of business logic.

The repository pattern puts one seam between your handlers and the ORM. Handlers and services depend on an interface — “give me the customer with this id”, “save this order” — and a SQLAlchemy implementation sits behind it. Swap in a fake for tests, swap the implementation if the persistence story changes, and keep the parts of the code that hold your business rules ignorant of how rows get written. This is a guide to doing that in FastAPI 0.115 with SQLAlchemy 2 and Python 3.12, and an honest note on when it is worth the ceremony and when it is not.

Define the interface, not the implementation

Start with a Protocol per aggregate. A protocol is structural typing — anything with the right methods satisfies it, no inheritance required — which is exactly what you want so that both the real repository and the fake conform without sharing a base class. The interface speaks the domain’s language, not the table’s.

# repositories/customer.py
from typing import Protocol
from uuid import UUID

from domain.models import Customer


class CustomerRepository(Protocol):
    """What the application needs of customer storage — nothing about SQL."""

    def get(self, customer_id: UUID) -> Customer | None: ...

    def get_by_email(self, email: str) -> Customer | None: ...

    def add(self, customer: Customer) -> None: ...

    def list_active(self, limit: int = 100) -> list[Customer]: ...

Note what is absent. No Session, no select(), no commit. The protocol is a contract your route handlers and services can be written against before a single line of SQLAlchemy exists. That is the whole point: the code that decides what happens stops depending on the code that decides how it is stored.

The SQLAlchemy implementation

The concrete repository holds a Session and is the only place that writes queries against the customer aggregate. SQLAlchemy 2 uses select() with session.scalars(...) rather than the legacy Query API, so the modern style reads cleanly.

# repositories/sqlalchemy/customer.py
from uuid import UUID

from sqlalchemy import select
from sqlalchemy.orm import Session

from domain.models import Customer


class SqlAlchemyCustomerRepository:
    def __init__(self, session: Session) -> None:
        self._session = session

    def get(self, customer_id: UUID) -> Customer | None:
        return self._session.get(Customer, customer_id)

    def get_by_email(self, email: str) -> Customer | None:
        stmt = select(Customer).where(Customer.email == email)
        return self._session.scalars(stmt).one_or_none()

    def add(self, customer: Customer) -> None:
        self._session.add(customer)

    def list_active(self, limit: int = 100) -> list[Customer]:
        stmt = (
            select(Customer)
            .where(Customer.is_active.is_(True))
            .order_by(Customer.created_at.desc())
            .limit(limit)
        )
        return list(self._session.scalars(stmt))

The repository deliberately does not commit. It reads and it stages writes. Deciding when a transaction commits is a different concern, and mixing it into the repository means every method has to guess whether it owns the transaction. That decision belongs one level up.

Unit of work: who owns the transaction

A single HTTP request that writes to two aggregates should commit once, atomically, or not at all. The unit-of-work pattern makes that explicit: it owns the session, exposes the repositories, and is the thing that commits or rolls back. In FastAPI you wire it through a dependency with yield, so the session is opened per request and closed after the response — the framework’s standard lifecycle for a resource.

# unit_of_work.py
from collections.abc import Iterator

from fastapi import Depends
from sqlalchemy.orm import Session, sessionmaker

from repositories.sqlalchemy.customer import SqlAlchemyCustomerRepository
from repositories.customer import CustomerRepository


class UnitOfWork:
    customers: CustomerRepository

    def __init__(self, session: Session) -> None:
        self._session = session
        self.customers = SqlAlchemyCustomerRepository(session)

    def commit(self) -> None:
        self._session.commit()

    def rollback(self) -> None:
        self._session.rollback()


def get_uow(session_factory: sessionmaker[Session]) -> Iterator[UnitOfWork]:
    session = session_factory()
    try:
        yield UnitOfWork(session)
    finally:
        session.close()

The route handler now reads at the level of intent. It never touches a Session and never writes a query. It asks the unit of work for what it needs and commits once at the end.

# routers/customers.py
from fastapi import APIRouter, Depends, HTTPException

from domain.models import Customer
from schemas import CustomerCreate, CustomerOut
from unit_of_work import UnitOfWork, get_uow

router = APIRouter(prefix="/customers")


@router.post("", response_model=CustomerOut, status_code=201)
def create_customer(
    payload: CustomerCreate,
    uow: UnitOfWork = Depends(get_uow),
) -> Customer:
    if uow.customers.get_by_email(payload.email) is not None:
        raise HTTPException(status_code=409, detail="Email already registered")
    customer = Customer.create(email=payload.email, name=payload.name)
    uow.customers.add(customer)
    uow.commit()
    return customer

The payoff: a fake repository in tests

This is where the abstraction earns its keep. Because handlers depend on the CustomerRepository protocol, a test can hand them a dictionary-backed fake and never open a database connection. The tests run in milliseconds, exercise real logic, and fail for real reasons rather than fixture plumbing.

# tests/fakes.py
from uuid import UUID

from domain.models import Customer


class FakeCustomerRepository:
    def __init__(self) -> None:
        self._store: dict[UUID, Customer] = {}

    def get(self, customer_id: UUID) -> Customer | None:
        return self._store.get(customer_id)

    def get_by_email(self, email: str) -> Customer | None:
        return next(
            (c for c in self._store.values() if c.email == email), None
        )

    def add(self, customer: Customer) -> None:
        self._store[customer.id] = customer

    def list_active(self, limit: int = 100) -> list[Customer]:
        active = [c for c in self._store.values() if c.is_active]
        return active[:limit]


# tests/test_customers.py
from fastapi.testclient import TestClient

from main import app
from unit_of_work import UnitOfWork, get_uow
from tests.fakes import FakeCustomerRepository


def test_duplicate_email_is_rejected() -> None:
    repo = FakeCustomerRepository()
    fake_uow = UnitOfWork.__new__(UnitOfWork)  # skip the Session
    fake_uow.customers = repo
    fake_uow.commit = lambda: None  # type: ignore[method-assign]

    app.dependency_overrides[get_uow] = lambda: fake_uow
    client = TestClient(app)

    first = client.post("/customers", json={"email": "a@b.com", "name": "A"})
    assert first.status_code == 201
    second = client.post("/customers", json={"email": "a@b.com", "name": "B"})
    assert second.status_code == 409

    app.dependency_overrides.clear()

FastAPI’s dependency_overrides is the supported hook for exactly this — replacing get_uow for the duration of a test and clearing it afterwards. The same discipline that keeps these tests fast is what keeps a CI pipeline honest; if you have not already got the basics wired up, a pre-commit baseline that actually runs is the place to put the type checks and linters that guard this seam.

The honest trade-off

The pattern is not free, and it is not always right. On a five-endpoint CRUD service that will never grow, a repository layer is ceremony over a small app — you have added indirection to hide a query that was never going to change. SQLAlchemy’s session is already a unit of work and an identity map; wrapping it in a thin passthrough that only ever has one implementation is abstraction for its own sake.

It earns its place when at least one of these is true: you have real domain logic worth testing without a database; your query logic is complex enough that concentrating it in one class stops it metastasising through the handlers; or you genuinely expect the persistence story to shift — read replicas, a cache in front of reads, a second store for a subset of aggregates. The tell is repeated, near-identical query code appearing in three route handlers, or a test suite that is slow because every case needs a live database. Both are forms of accumulating drag on delivery that the pattern is designed to pay down.

Introduce it per aggregate, not as a big-bang refactor. Wrap the one aggregate whose queries are sprawling or whose logic you keep wanting to test in isolation, prove the seam with a fake, and leave the rest talking to the session directly until they hurt. The same restraint applies to the layers you build alongside it — a repository is worth adding for the same reason idempotency handling is: because you have a concrete problem it solves, not because the architecture diagram looks tidier with another box in it.

Depend on an interface where change is plausible; talk to the ORM directly where it is not. The repository pattern is a tool for buying optionality, and optionality you will never exercise is just cost.