SQLAlchemy Cheatsheet 12 — Repository and Unit-of-Work Pattern

Cheatsheet for repository / UoW patterns. Where you’d use them, where you wouldn’t.

When to use a repository

You want to test handlers without a DB.
Persistence may change (DB → ES → API).
The team needs a strict boundary between domain and persistence.

When not:

Small CRUD apps — Session is fine as a “repository.”
Tight coupling to SQL features is OK.

Basic repository

class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def get(self, id: int) -> User | None:
        return await self.session.get(User, id)
    
    async def by_email(self, email: str) -> User | None:
        return await self.session.scalar(select(User).where(User.email == email))
    
    async def list_(self, limit: int = 20, cursor: int = 0) -> list[User]:
        stmt = (
            select(User)
            .where(User.id > cursor)
            .order_by(User.id)
            .limit(limit)
        )
        return list((await self.session.execute(stmt)).scalars())
    
    async def add(self, user: User) -> User:
        self.session.add(user)
        await self.session.flush()
        return user
    
    async def delete(self, id: int) -> None:
        user = await self.session.get(User, id)
        if user:
            await self.session.delete(user)

DI in FastAPI

async def get_user_repo(db: AsyncSession = Depends(get_db)) -> UserRepository:
    return UserRepository(db)

@app.get("/users/{id}")
async def get_user(id: int, repo: UserRepository = Depends(get_user_repo)):
    user = await repo.get(id)
    if not user:
        raise HTTPException(404)
    return user

Generic base repo

from typing import TypeVar, Generic, Type

T = TypeVar("T", bound=Base)

class BaseRepository(Generic[T]):
    model: Type[T]
    
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def get(self, id: int) -> T | None:
        return await self.session.get(self.model, id)
    
    async def add(self, obj: T) -> T:
        self.session.add(obj)
        await self.session.flush()
        return obj
    
    async def delete(self, id: int) -> None:
        obj = await self.get(id)
        if obj:
            await self.session.delete(obj)
    
    async def list_(self, *, limit: int = 20, **filters) -> list[T]:
        stmt = select(self.model).limit(limit)
        for k, v in filters.items():
            stmt = stmt.where(getattr(self.model, k) == v)
        return list((await self.session.execute(stmt)).scalars())

class UserRepository(BaseRepository[User]):
    model = User
    
    async def by_email(self, email: str) -> User | None:
        return await self.session.scalar(select(User).where(User.email == email))

Unit of Work

class UnitOfWork:
    def __init__(self, session_factory):
        self.session_factory = session_factory
    
    async def __aenter__(self):
        self.session = self.session_factory()
        self.users = UserRepository(self.session)
        self.posts = PostRepository(self.session)
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        if exc:
            await self.session.rollback()
        else:
            await self.session.commit()
        await self.session.close()

# Usage
async with UnitOfWork(AsyncSessionLocal) as uow:
    user = User(email="[email protected]")
    await uow.users.add(user)
    post = Post(author_id=user.id, title="hi")
    await uow.posts.add(post)
# auto-commit on exit

Service layer (business logic)

class UserService:
    def __init__(self, uow: UnitOfWork):
        self.uow = uow
    
    async def register(self, data: UserCreate) -> User:
        async with self.uow as uow:
            if await uow.users.by_email(data.email):
                raise EmailTaken
            user = User(email=data.email, hashed_password=hash_pw(data.password))
            await uow.users.add(user)
            return user

async def get_uow() -> UnitOfWork:
    return UnitOfWork(AsyncSessionLocal)

async def get_user_service(uow: UnitOfWork = Depends(get_uow)) -> UserService:
    return UserService(uow)

@app.post("/users", response_model=UserOut)
async def register(data: UserCreate, svc: UserService = Depends(get_user_service)):
    return await svc.register(data)

Testing repos

@pytest.fixture
async def repo(db_session):
    return UserRepository(db_session)

async def test_create(repo, db_session):
    u = User(email="[email protected]")
    await repo.add(u)
    await db_session.commit()
    fetched = await repo.get(u.id)
    assert fetched.email == "[email protected]"

Faked repo in service tests

class FakeUserRepo:
    def __init__(self):
        self.store = {}
    async def by_email(self, email):
        return next((u for u in self.store.values() if u.email == email), None)
    async def add(self, user):
        user.id = len(self.store) + 1
        self.store[user.id] = user
        return user

Tradeoffs

	Repository	Direct Session
Testability	Easier (fake repo)	Harder (real DB)
Coupling	Lower	Higher
Boilerplate	More	Less
Use of SQL features	Wrapped	Direct
Team size	Helpful at scale	Lean for small

For most FastAPI apps: Session-as-repository is fine. Reach for the pattern when boundaries matter.

Common mistakes

Generic repo with too-generic methods that hide intent.
Repo that exposes the session — defeats the purpose.
Service layer that’s just a passthrough — adds tax without value.
UoW that doesn’t actually unify transactions.

When to use a repository#

Basic repository#

DI in FastAPI#

Generic base repo#

Unit of Work#

Service layer (business logic)#

Testing repos#

Faked repo in service tests#

Tradeoffs#

Common mistakes#

Read this next#

When to use a repository

Basic repository

DI in FastAPI

Generic base repo

Unit of Work

Service layer (business logic)

Testing repos

Faked repo in service tests

Tradeoffs

Common mistakes

Read this next