Repository and Unit of Work Patterns

Two complementary persistence patterns popularized by Martin Fowler’s PoEAA (2002). Repository abstracts data access; Unit of Work tracks changes and commits them as a single transaction. Most modern ORMs (SQLAlchemy, Django ORM, Hibernate) implement Unit of Work; explicit Repositories are often layered on top.

For DDD context see 12_domain_driven_design.md. For Clean Architecture see 09_clean_architecture.md.

Repository

A collection-like abstraction over persistence. The domain talks to repositories as if they were in-memory collections.

class OrderRepository(Protocol):
    def get(self, id: OrderId) -> Order | None: ...
    def find_by_customer(self, customer_id: CustomerId) -> list[Order]: ...
    def save(self, order: Order) -> None: ...
    def delete(self, order: Order) -> None: ...

To the domain, this looks like a smart list. Underneath it’s SQL, ORM, or whatever:

class SqlAlchemyOrderRepository:
    def __init__(self, session):
        self.session = session

    def get(self, id):
        row = self.session.get(OrderModel, id)
        return row.to_domain() if row else None

    def find_by_customer(self, customer_id):
        rows = self.session.query(OrderModel).filter_by(customer_id=customer_id).all()
        return [r.to_domain() for r in rows]

    def save(self, order):
        model = OrderModel.from_domain(order)
        self.session.merge(model)

    def delete(self, order):
        self.session.delete(self.session.get(OrderModel, order.id))

The interface is in the domain (or application) layer. The implementation lives in infrastructure. Domain code depends on the interface; tests use a fake.

Why repository

Fake repository for tests

class FakeOrderRepository:
    def __init__(self):
        self.orders: dict[OrderId, Order] = {}

    def get(self, id):
        return self.orders.get(id)

    def find_by_customer(self, customer_id):
        return [o for o in self.orders.values() if o.customer_id == customer_id]

    def save(self, order):
        self.orders[order.id] = order

    def delete(self, order):
        del self.orders[order.id]

Pure Python, no DB. Tests of use cases that depend on the repository run in milliseconds.

Repository per aggregate root (DDD)

In DDD, the rule is one repository per aggregate root. Not per table, not per entity, not per query.

class OrderRepository:        # OK — Order is an aggregate root
    def get(self, id): ...
    def save(self, order): ...

class OrderItemRepository:    # WRONG — OrderItem is part of Order's aggregate
    ...

Loading an Order brings its OrderItems along. Persisting an Order persists its items. The aggregate is the unit; the repository operates on units.

When the repository pattern becomes ceremony

class UserRepository:
    def get_by_id(self, id):
        return User.objects.get(id=id)
    def get_by_email(self, email):
        return User.objects.get(email=email)
    def save(self, user):
        user.save()

For simple CRUD with Django ORM (or SQLAlchemy), the ORM IS the repository. Wrapping it adds boilerplate. Consider whether the abstraction is paying off.

The repository pays off when:

• You have rich domain entities separate from ORM models.
• Tests need to swap implementations.
• Query logic is complex enough to centralize.

It doesn’t pay off when:

• Entities are essentially ORM models.
• Queries are one-liners.
• You’re using Django Admin / similar that needs ORM directly.

Unit of Work

Tracks a set of changes; commits or rolls back as one atomic operation.

class UnitOfWork:
    def __init__(self, session_factory):
        self.session_factory = session_factory

    def __enter__(self):
        self.session = self.session_factory()
        self.orders = SqlAlchemyOrderRepository(self.session)
        self.customers = SqlAlchemyCustomerRepository(self.session)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.session.rollback()
        else:
            self.session.commit()
        self.session.close()

# Usage
with UnitOfWork(SessionFactory) as uow:
    order = uow.orders.get(order_id)
    customer = uow.customers.get(order.customer_id)
    order.apply_loyalty_discount(customer.tier)
    uow.orders.save(order)
    # commit happens on __exit__

Two repositories share one DB session. Changes are committed atomically.

Why Unit of Work

• Transactional boundaries are explicit: code reads “everything in this block is atomic.”
• Multiple repositories, one transaction: cross-aggregate operations.
• Rollback on failure: any exception undoes all changes.
• Performance: batched writes, fewer round trips.

SQLAlchemy’s session IS a Unit of Work

with Session(engine) as session:
    user = session.get(User, 1)
    user.email = "new@example.com"
    order = Order(customer_id=1, ...)
    session.add(order)
    # session.commit() — flushes user update + order insert as one transaction

The session tracks loaded objects, detects changes, batches writes, manages transactions. You don’t need to build a Unit of Work — SQLAlchemy is one.

For Django, the transaction.atomic() context manager is the rough equivalent:

with transaction.atomic():
    user.email = "new@example.com"
    user.save()
    Order.objects.create(customer=user, ...)

Custom Unit of Work over the ORM session

The pattern of wrapping the session in a custom UoW class makes sense when:

• You want to hide SQLAlchemy from the domain (DDD-style).
• Multiple repositories need coordinated access to one session.
• You want a uniform API across backends.

class UnitOfWork(Protocol):
    orders: OrderRepository
    customers: CustomerRepository

    def __enter__(self) -> "UnitOfWork": ...
    def __exit__(self, *args) -> None: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...

The use case takes a UoW:

class PlaceOrderUseCase:
    def __init__(self, uow: UnitOfWork):
        self.uow = uow

    def execute(self, customer_id, items):
        with self.uow:
            customer = self.uow.customers.get(customer_id)
            order = Order.create(customer, items)
            self.uow.orders.save(order)
        return order

Tests use a FakeUnitOfWork:

class FakeUnitOfWork:
    def __init__(self):
        self.orders = FakeOrderRepository()
        self.customers = FakeCustomerRepository()
        self.committed = False

    def __enter__(self): return self
    def __exit__(self, *args): pass
    def commit(self): self.committed = True
    def rollback(self): pass

Aggregate-aware persistence

The repository should save the entire aggregate atomically:

class OrderRepository:
    def save(self, order: Order):
        # Save Order
        order_row = OrderModel.from_domain(order)
        self.session.merge(order_row)
        # Save its OrderItems — they're part of the aggregate
        self.session.query(OrderItemModel).filter_by(order_id=order.id).delete()
        for item in order.items:
            self.session.add(OrderItemModel.from_domain(item, order.id))

Or with cascade configured in the ORM, just save the root:

class OrderModel(Base):
    __tablename__ = "orders"
    items = relationship("OrderItemModel", cascade="all, delete-orphan")

Now session.add(order_row) cascades to items. The repository looks simpler.

Read-only repositories (CQRS-ish)

For complex queries that don’t fit the “load an aggregate” pattern, use a separate read-side repository:

class OrderRepository:           # write side — works on aggregates
    def get(self, id): ...
    def save(self, order): ...

class OrderQueries:               # read side — denormalized, query-optimized
    def list_for_customer_dashboard(self, customer_id) -> list[OrderSummary]:
        return self.session.execute(
            "SELECT order_id, total, status, item_count "
            "FROM order_summaries "
            "WHERE customer_id = %s",
            customer_id,
        )

The write repository loads/saves aggregates. The read queries return DTOs / dicts for UI consumption. No need to load full aggregates for read-only views.

This is light CQRS. See 13_cqrs_event_sourcing.md.

Specification pattern (for complex queries)

When queries get complex and dynamic, encapsulate them:

class OrdersBy:
    @staticmethod
    def customer(customer_id):
        return lambda q: q.filter(Order.customer_id == customer_id)

    @staticmethod
    def status(status):
        return lambda q: q.filter(Order.status == status)

    @staticmethod
    def created_after(date):
        return lambda q: q.filter(Order.created_at > date)

# Usage
orders = repo.find(OrdersBy.customer(42), OrdersBy.status("paid"), OrdersBy.created_after(cutoff))

Or via objects:

class CustomerOrdersSpec:
    def __init__(self, customer_id, status=None):
        self.customer_id = customer_id
        self.status = status

    def to_query(self, base_query):
        q = base_query.filter(Order.customer_id == self.customer_id)
        if self.status:
            q = q.filter(Order.status == self.status)
        return q

orders = repo.find(CustomerOrdersSpec(customer_id=42, status="paid"))

Useful for queries that vary by user input. Avoids per-combination method explosion.

Common pitfalls

• One method per query in the repository: find_active_users_in_california_aged_25_to_35 — combinatorial. Use specifications or filter args.
• Repository leaking ORM types: methods return SQLAlchemy Query objects or Django querysets. Domain code now depends on the ORM. Return domain objects (or DTOs).
• Repository depending on the framework: importing Flask request in a repository. Repository should know nothing of HTTP.
• Cross-aggregate Unit of Work spanning days: holding a session open through user idle time. Sessions should be request-scoped (or unit-of-work scoped).
• Calling commit mid-operation: defeats atomicity. One commit per logical unit.
• N+1 in repository methods: find_all_users() returns users; iterating to load related data hits the DB per user. Eager-load in the query.

Common interview confusions

• “Repository is just a DAO.” — historically yes (DAO = Data Access Object, similar idea). DDD’s repository emphasizes the “collection of aggregates” mental model; DAO is more “table-row mapper.” Different vocabularies, overlapping concepts.
• “You always need a repository.” — for simple CRUD with Django/SQLAlchemy, the ORM is the repository. Extra wrapping is boilerplate. Add a repository when you have rich domain entities to translate to/from.
• “Unit of Work means using transactions.” — every DB transaction is a kind of UoW. The PATTERN means tracking changes across multiple repositories and committing as one — a concrete code construct (the UoW class).

Interview angle

• “What’s the Repository pattern?” — abstraction over persistence; treats the data store as a collection of domain objects. OrderRepository.get(id), OrderRepository.save(order). Domain code depends on the interface; concrete impl (SQLAlchemy, Mongo) lives in infrastructure. Tests use fakes.
• “When is the Repository pattern overkill?” — simple CRUD with Django/SQLAlchemy where the ORM IS effectively the repository. Wrapping User.objects.get(id=id) in UserRepository.get_by_id(id) is boilerplate.
• “Repository per aggregate vs per table?” — in DDD, repository per aggregate root. The aggregate is the consistency boundary; the repository loads/saves it as a unit (including child entities).
• “What is the Unit of Work pattern?” — tracks a set of changes across multiple repositories and commits or rolls back as one transaction. SQLAlchemy session is a UoW implementation. Useful for cross-aggregate operations where atomicity matters.
• “How does SQLAlchemy implement Unit of Work?” — the session tracks loaded objects, dirty changes, new additions, deletions. On commit(), it flushes all changes as one transaction. Rollback discards everything since last commit.
• “Why would you build a custom Unit of Work class on top of SQLAlchemy?” — to hide SQLAlchemy from the domain (DDD-style), to coordinate multiple repositories under one session, to provide a swappable interface for tests. Common in clean-architecture codebases.
• “What’s the specification pattern?” — encapsulates complex query criteria as objects. Each spec describes part of the filter; specs combine. Useful when queries vary dynamically and method-per-query becomes combinatorial.
• “What’s a fake repository?” — in-memory implementation of the repository interface, used in tests. No DB; fast and deterministic. Better than mocks because it behaves consistently across tests.