Skip to content

Repository

Intent

Give domain code a collection-like object (add, get, find_by_email, remove) that hides where entities live and how they are fetched. The service that registers a user never sees SQL, connections or row mapping; it sees a contract, and the same service runs over an in-memory dict in tests and over a database in production.

When to use and when not to

Use it when

  • Your LLD problem has entities that outlive a request (users, tickets, orders, accounts) and the interviewer will ask "where does this persist?"; the repository is the one-line answer and the seam where a database plugs in later.
  • You want tests of business rules that need no database: the dict-backed fake is the reason the pattern exists.
  • Several callers need the same queries (find_by_email, find_overdue); the repository names each query once and keeps query logic out of services.
  • Storage exceptions must become domain exceptions (ConflictError, NotFoundError) so callers branch on meaning, not on driver classes.

Leave it out when

  • The state is process-local and short-lived (a cache, a game board, a parser's stack): a dict or a list is the design, not a missing abstraction.
  • There is one entity, one get and nothing to protect with tests; wrapping a dict in a class buys nothing yet.
  • An ORM session already is the repository (Django managers, SQLAlchemy Session); add your own layer only to narrow the interface, never to mirror it.
  • The query is a report across many entity types with joins; repositories work per entity, a read model or query object works for reporting.

Structure

Four roles: the entity, the collection-like contract, one implementation per storage technology, and the service that depends on the contract alone.

classDiagram
    class User {
        +id: str
        +email: str
        +name: str
    }
    class UserRepository {
        <<interface>>
        +add(user: User)
        +get(user_id: str) User
        +find_by_email(email: str) User
        +update(user: User)
        +remove(user_id: str)
        +count() int
    }
    class InMemoryUserRepository {
        -_by_id: Dict~str,User~
        -_by_email: Dict~str,str~
        -_lock: Lock
    }
    class SqliteUserRepository {
        -_conn: Connection
    }
    class RegistrationService {
        -_users: UserRepository
        -_ids: IdGenerator
        +register(email: str, name: str) User
        +rename(user_id: str, name: str) User
    }
    UserRepository <|.. InMemoryUserRepository
    UserRepository <|.. SqliteUserRepository
    RegistrationService o-- UserRepository : depends on
    UserRepository ..> User : stores

RegistrationService holds a UserRepository and cannot tell the implementations apart. The dotted arrows are realisation: both classes qualify by shape. The contract includes the errors each method may raise, which is what makes the implementations interchangeable rather than merely similar.

Canonical example in Python

Entity and contract first (code/patterns/repository.py, tested by code/patterns/tests/test_repository.py):

code/patterns/repository.py — the entity and the contract
@dataclass(frozen=True, slots=True)
class User:
    """The entity the repository stores. Frozen: a change is a new value passed to ``update``.

    Frozen matters for the in-memory implementation. If ``get`` handed out a mutable
    object, a caller could change it without calling ``update`` and the fake would
    "persist" the change while the SQL implementation would not. Value semantics keep
    both implementations honest in the same way.
    """

    id: str
    email: str
    name: str


@runtime_checkable
class UserRepository(Protocol):
    """The collection-like contract. Domain code sees these six methods and nothing else.

    The contract includes the errors: ``add`` raises ``ConflictError`` when the id or
    the email is already taken; ``update`` and ``remove`` raise ``NotFoundError`` when
    the user does not exist. Every implementation must agree, and the contract tests
    in ``tests/test_repository.py`` run against all of them to prove it.
    """

    def add(self, user: User) -> None: ...

    def get(self, user_id: str) -> User | None: ...

    def find_by_email(self, email: str) -> User | None: ...

    def update(self, user: User) -> None: ...

    def remove(self, user_id: str) -> None: ...

    def count(self) -> int: ...

Three decisions to say out loud:

  • The contract names the errors. add raises ConflictError for a taken id or email; update and remove raise NotFoundError. An interface that lists only method names lets each implementation invent its own failure mode, and the fake then diverges from production exactly where the tests matter.
  • The entity is a frozen value. If get handed out a mutable object, the in-memory fake would "persist" a caller's mutation without update, while the SQL implementation would not. Frozen entities force replace plus update on both, so a test that passes on the fake passes on the database.
  • Queries are methods, not filters. find_by_email is a named query with a stable meaning; a filter(**kwargs) escape hatch leaks the storage schema into every caller.

The in-memory implementation is the test fake and a legitimate small-scale store. The lock makes the uniqueness check plus the insert one atomic step, the way a UNIQUE index does for the SQL version:

code/patterns/repository.py — the dict-backed implementation
class InMemoryUserRepository:
    """A dict-backed implementation: the test fake, and a legitimate small-scale store.

    ``_lock`` protects ``_by_id`` and ``_by_email`` together, so the uniqueness check
    and the insert are one atomic step: exactly what the UNIQUE index gives the SQL
    implementation. Without it two threads could both pass the check and both insert.
    """

    def __init__(self) -> None:
        self._by_id: dict[str, User] = {}
        self._by_email: dict[str, str] = {}  # email -> user id: the "unique index"
        self._lock = threading.Lock()

    def add(self, user: User) -> None:
        with self._lock:
            if user.id in self._by_id:
                raise ConflictError(f"user {user.id} already exists")
            if user.email in self._by_email:
                raise ConflictError(f"email {user.email} is already registered")
            self._by_id[user.id] = user
            self._by_email[user.email] = user.id

    def get(self, user_id: str) -> User | None:
        with self._lock:
            return self._by_id.get(user_id)

    def find_by_email(self, email: str) -> User | None:
        with self._lock:
            user_id = self._by_email.get(email)
            return None if user_id is None else self._by_id[user_id]

    def update(self, user: User) -> None:
        with self._lock:
            current = self._by_id.get(user.id)
            if current is None:
                raise NotFoundError(f"user {user.id} does not exist")
            owner = self._by_email.get(user.email)
            if owner is not None and owner != user.id:
                raise ConflictError(f"email {user.email} is already registered")
            del self._by_email[current.email]
            self._by_email[user.email] = user.id
            self._by_id[user.id] = user

    def remove(self, user_id: str) -> None:
        with self._lock:
            user = self._by_id.pop(user_id, None)
            if user is None:
                raise NotFoundError(f"user {user_id} does not exist")
            del self._by_email[user.email]

    def count(self) -> int:
        with self._lock:
            return len(self._by_id)

The SQLite implementation translates sqlite3.IntegrityError into ConflictError and uses rowcount to detect a missing row. The connection is injected, so whoever owns the transaction decides when the writes become visible; that owner is the Unit of Work.

code/patterns/repository.py — the sqlite3 implementation
SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
    id    TEXT PRIMARY KEY,
    email TEXT NOT NULL UNIQUE,
    name  TEXT NOT NULL
)
"""


def connect(path: str = ":memory:") -> sqlite3.Connection:
    """Open a connection with the schema applied; ``:memory:`` gives every test a fresh database.

    ``autocommit=True`` makes each statement its own transaction, which is what a
    repository used on its own wants. A Unit of Work takes over that decision.
    """
    conn = sqlite3.connect(path, autocommit=True)
    conn.execute(SCHEMA)
    return conn


class SqliteUserRepository:
    """The same contract over a relational table.

    The connection is injected, so whoever owns the transaction decides when the
    writes become visible. Storage exceptions are translated into the domain errors
    the contract promises; callers never import ``sqlite3``. One connection serves
    one thread (``sqlite3``'s default ``check_same_thread``).
    """

    def __init__(self, conn: sqlite3.Connection) -> None:
        self._conn = conn

    def add(self, user: User) -> None:
        try:
            self._conn.execute(
                "INSERT INTO users (id, email, name) VALUES (?, ?, ?)",
                (user.id, user.email, user.name),
            )
        except sqlite3.IntegrityError as exc:
            raise ConflictError(f"user {user.id} or email {user.email} already exists") from exc

    def get(self, user_id: str) -> User | None:
        row = self._conn.execute(
            "SELECT id, email, name FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        return _to_user(row)

    def find_by_email(self, email: str) -> User | None:
        row = self._conn.execute(
            "SELECT id, email, name FROM users WHERE email = ?", (email,)
        ).fetchone()
        return _to_user(row)

    def update(self, user: User) -> None:
        try:
            cursor = self._conn.execute(
                "UPDATE users SET email = ?, name = ? WHERE id = ?",
                (user.email, user.name, user.id),
            )
        except sqlite3.IntegrityError as exc:
            raise ConflictError(f"email {user.email} is already registered") from exc
        if cursor.rowcount == 0:
            raise NotFoundError(f"user {user.id} does not exist")

    def remove(self, user_id: str) -> None:
        cursor = self._conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
        if cursor.rowcount == 0:
            raise NotFoundError(f"user {user_id} does not exist")

    def count(self) -> int:
        (count,) = self._conn.execute("SELECT COUNT(*) FROM users").fetchone()
        return int(count)


def _to_user(row: tuple[str, str, str] | None) -> User | None:
    """The data mapper: one row of the ``users`` table becomes one entity."""
    return None if row is None else User(id=row[0], email=row[1], name=row[2])

The service depends on the contract alone. Its email check gives a friendly error, but the repository's add is the authoritative guard: two registrations can pass the check at the same moment, and only the repository (its lock, or its index) sees both.

code/patterns/repository.py — the domain service
class RegistrationService:
    """Domain code: depends on the contract, never on a storage technology.

    The email check here produces a friendly error; the repository's ``add`` is the
    authoritative guard, because two registrations can pass the check at the same
    moment and only the repository (its lock, or its UNIQUE index) sees both.
    """

    def __init__(self, users: UserRepository, ids: IdGenerator) -> None:
        self._users = users
        self._ids = ids

    def register(self, email: str, name: str) -> User:
        email = email.strip().lower()
        if "@" not in email:
            raise ValidationError(f"{email!r} is not an email address")
        if self._users.find_by_email(email) is not None:
            raise ConflictError(f"email {email} is already registered")
        user = User(id=self._ids.next_id(), email=email, name=name)
        self._users.add(user)
        return user

    def rename(self, user_id: str, name: str) -> User:
        user = self._users.get(user_id)
        if user is None:
            raise NotFoundError(f"user {user_id} does not exist")
        renamed = replace(user, name=name)
        self._users.update(renamed)
        return renamed

Running python -m patterns.repository prints:

--- registration over the in-memory repository ---
registered user-1 as ada@example.com
rejected: email ada@example.com is already registered
renamed user-1 to 'Ada Lovelace'; 1 user(s) stored
rejected: user user-99 does not exist
--- registration over the sqlite repository ---
registered user-1 as ada@example.com
rejected: email ada@example.com is already registered
renamed user-1 to 'Ada Lovelace'; 1 user(s) stored
rejected: user user-99 does not exist
--- generic dict-backed repository, keyed by a function ---
2 books; isinstance(books, Repository) is True
  978-0201633610: Design Patterns
  978-0321127426: Patterns of Enterprise Application Architecture

Pythonic variant

Python 3.12 lets you write the shared shape once as a generic Protocol and build a single dict-backed implementation for every entity type:

code/patterns/repository.py — Repository[T] and one fake for all entities
@runtime_checkable
class Repository[T](Protocol):
    """The shape every repository shares; ``T`` is the entity type (PEP 695 syntax)."""

    def add(self, item: T) -> None: ...

    def get(self, item_id: str) -> T | None: ...

    def remove(self, item_id: str) -> None: ...

    def __iter__(self) -> Iterator[T]: ...

    def __len__(self) -> int: ...


class InMemoryRepository[T]:
    """One dict-backed implementation for every entity type.

    ``InMemoryRepository[Book](key=lambda book: book.isbn)`` is the fake you write once
    and reuse across an LLD problem's tests. The entity-specific classes above exist
    because ``find_by_email`` and the uniqueness rule belong to users specifically.
    ``_lock`` protects ``_items``.
    """

    def __init__(self, key: Callable[[T], str]) -> None:
        self._key = key
        self._items: dict[str, T] = {}
        self._lock = threading.Lock()

    def add(self, item: T) -> None:
        item_id = self._key(item)
        with self._lock:
            if item_id in self._items:
                raise ConflictError(f"{item_id} already exists")
            self._items[item_id] = item

    def get(self, item_id: str) -> T | None:
        with self._lock:
            return self._items.get(item_id)

    def remove(self, item_id: str) -> None:
        with self._lock:
            if self._items.pop(item_id, None) is None:
                raise NotFoundError(f"{item_id} does not exist")

    def __iter__(self) -> Iterator[T]:
        with self._lock:
            return iter(list(self._items.values()))

    def __len__(self) -> int:
        with self._lock:
            return len(self._items)


@dataclass(frozen=True, slots=True)
class Book:
    """A second entity type, identified by ISBN rather than ``id``: the key function copes."""

    isbn: str
    title: str
  • Repository[T] in PEP 695 syntax. class Repository[T](Protocol) declares the type parameter inline; InMemoryRepository[Book](key=lambda book: book.isbn) reads like a typed collection.
  • A key function instead of a base class. Entities need not inherit from Entity or expose .id; the repository is told how to identify them.
  • __iter__ and __len__ make the repository feel like the collection it stands in for: for user in users, len(users), any(...).

When is the generic fake enough? For services whose needs are add, get and remove. The moment a service needs find_by_email with a uniqueness rule, write the entity-specific Protocol: the query and the invariant are part of the contract, and production must enforce them too.

Reach for When
A plain dict Process-local state with no persistence story
InMemoryRepository[T] Tests and demos for any entity that only needs add, get and remove
An entity-specific Protocol such as UserRepository Named queries and uniqueness rules that every implementation must honour
A SQL implementation behind the same Protocol The interviewer asks "and in production?"

Real-world usage

  • Django managers and QuerySets. User.objects.get(pk=...) is a repository attached to the model class; a custom manager such as Article.published is a named query.
  • SQLAlchemy Session. session.add(obj) and session.get(User, id) make the session a repository plus a unit of work. Cosmic Python (Percival and Gregory) puts an explicit AbstractRepository in front of it so the domain never imports the ORM; that is the shape used here.
  • Standard library. shelve and dbm give dict-shaped persistence; collections.abc.MutableMapping is the collection contract a repository imitates.
Looks like Repository How to tell them apart
DAO (Data Access Object) A DAO mirrors a table (insert_row, update_row) and exposes storage concepts; a repository mirrors a collection of domain objects and hides them. If the method names contain SQL verbs, you wrote a DAO.
Unit of Work The repository says what is stored; the unit of work says when the writes become visible, together. A repository that commits inside add cannot take part in a multi-repository transaction.
Active Record The entity saves itself (user.save()); a repository keeps persistence out of the entity. Django models are Active Record with a repository (the manager) attached.
Data Mapper The code that turns rows into objects; _to_user is a small one. A repository uses a mapper, it is not one.
Cache A cache may forget; a repository may not. A cache in front of a repository is a Proxy, not a second repository.
Dependency Injection DI is how the chosen repository reaches the service; the repository is what gets injected.

Where it appears in LLD problems

Interview tips

Interview tip

Introduce it as a sentence, not a diagram: "persistence sits behind a UserRepository Protocol; I implement it with a dict now, and the same tests prove a SQL version later." Then add the two things that mark an SDE2: the domain errors belong in the contract, and a contract test runs against every implementation.

Common mistake

A repository that leaks its storage: returning cursors or ORM rows, exposing filter(**kwargs), letting sqlite3.IntegrityError reach callers, or growing a save() that silently upserts. Runner-up: a fake that is friendlier than production (accepts duplicates, hands out shared mutable objects), so the tests pass and the database raises.