Architecture Patterns repository-pattern data-access-layer domain-driven-design Oleksandr Andrushchenko Published on — Modified on

What Is the Repository Pattern? Data Access Abstraction and Practical Examples

0 0

The Repository Pattern is a software design pattern that separates domain and application logic from the details of data persistence. It provides a collection-like interface for retrieving and storing domain objects, while hiding how those objects are mapped to database records. Instead of writing SQL queries or manipulating ORM sessions directly inside business services, application code works through repository operations defined around its domain model.

Described in Martin Fowler's Patterns of Enterprise Application Architecture, the pattern is widely used with Domain-Driven Design, Hexagonal Architecture, Onion Architecture, and Clean Architecture. But a repository is not automatically useful just because an application has a database. This guide explains its responsibilities, how it differs from a DAO and an ORM, how to handle transactions and concurrency, and how to implement a practical repository using Python, FastAPI, PostgreSQL, SQLAlchemy, and pytest.

Table of Contents

What Is the Repository Pattern?

The Repository Pattern provides an abstraction between the application's domain model and its persistence mechanism. To application code, a repository behaves like a collection of domain objects: it can find existing objects, add new ones, and persist changes. Behind that interface, a concrete implementation performs database operations and converts stored records into domain objects.

The pattern is particularly useful when an application has a meaningful domain model, complex queries, or multiple use cases that require consistent persistence behavior. Its purpose is to protect application logic from unnecessary database details—not to hide every database capability behind a generic CRUD interface.

The Problem It Solves

Consider a Python backend where the order service directly uses SQLAlchemy:

from sqlalchemy.orm import Session

from app.database import OrderRow


def get_customer_order(
    session: Session,
    order_id: str,
    customer_id: str,
):
    return (
        session.query(OrderRow)
        .filter(
            OrderRow.order_id == order_id,
            OrderRow.customer_id == customer_id,
        )
        .one_or_none()
    )

This approach is valid for a small application. However, as business behavior grows, services may accumulate query construction, persistence mapping, transaction control, and application rules in the same modules.

Several problems can emerge:

  • Multiple services duplicate the same query and filtering logic.
  • Use cases depend directly on ORM models and sessions.
  • Tests require mocking database details or setting up database infrastructure.
  • Persistence-specific exceptions leak into application behavior.
  • Domain objects become tightly coupled to database schemas.
  • Changing query or storage implementation requires editing business-oriented code.

A repository introduces a stable boundary. Application services can request an order by ID without knowing how the query is constructed or how the stored representation is mapped.

Core Responsibilities of a Repository

A repository should provide a focused interface for accessing and persisting domain objects.

Typical responsibilities include:

  • Retrieving aggregates using domain identifiers.
  • Persisting newly created aggregates.
  • Saving changes while preserving required concurrency guarantees.
  • Encapsulating reusable queries meaningful to the application.
  • Mapping database records into domain objects.
  • Translating persistence conflicts into application-meaningful outcomes.

For example, find_pending_orders_for_customer() expresses a business-oriented query, while execute_sql() exposes a lower-level mechanism.

A repository is not normally responsible for sending emails, authorizing payments, or deciding whether an order can be cancelled. Those responsibilities belong to appropriate domain or application components.

How the Repository Pattern Works

A typical implementation contains a repository contract and one or more concrete implementations. Application use cases depend on the contract, while infrastructure provides the actual persistence implementation.

This arrangement allows business code to remain independent of SQLAlchemy, PostgreSQL, and other persistence details where that independence is valuable.

Domain-Oriented Repository Interfaces

The repository interface should reflect the operations the application needs, using domain terminology and types.

from typing import Protocol

from app.domain import Order


class OrderRepository(Protocol):
    def get_by_id(
        self,
        order_id: str,
    ) -> Order | None:
        ...

    def add(self, order: Order) -> bool:
        ...

    def save(
        self,
        order: Order,
        expected_version: int,
    ) -> bool:
        ...

This interface works with an Order domain object rather than a SQLAlchemy row. It also describes the concurrency contract of updates through an expected version.

The distinction matters because the application needs to know whether an update succeeded or conflicted, but it does not need to know how a PostgreSQL UPDATE statement implements that condition.

Concrete Database and In-Memory Implementations

A production implementation may use PostgreSQL and SQLAlchemy. A unit test can use a small in-memory implementation instead.

Both must obey the same observable contract. For instance, if add() returns False when an order already exists, the in-memory implementation must behave that way too.

A fake repository helps tests focus on use-case behavior. It does not replace integration tests, because an ordinary Python dictionary does not reproduce database locking, transaction isolation, SQL behavior, or connection failures.

Other implementations are possible, but database replacements should not be assumed to preserve identical semantics. An abstraction cannot make PostgreSQL transactions and DynamoDB consistency models interchangeable without design work.

Repository vs. DAO vs. ORM

The Repository Pattern is often confused with Data Access Objects and Object-Relational Mappers. These concepts overlap, but their responsibilities differ.

Concept Primary Responsibility Typical Interface
Repository Provide access to domain objects and aggregates orders.get_by_id(id)
DAO Encapsulate database access and operations order_dao.select_by_id(id)
ORM Map objects to relational database representations session.get(OrderRow, id)
Unit of Work Coordinate changes in one transaction uow.commit()

A repository may use a DAO or ORM internally. SQLAlchemy can already provide some repository-like functionality, and adding another layer is unnecessary unless it establishes a useful application boundary.

A useful repository exposes domain intentions and consistency requirements instead of simply renaming every ORM operation.

Designing Repository Boundaries

The most important repository design decision is choosing what the interface represents. A repository can become a useful domain abstraction or a complicated wrapper that hides important storage behavior.

Design around business use cases and aggregate consistency rather than generating the same interface for every database table.

Repositories and Aggregate Roots

In Domain-Driven Design, repositories commonly operate on aggregate roots. An aggregate groups domain objects that must preserve consistency invariants through a controlled root.

For example, an Order aggregate may contain multiple OrderItems. Application code retrieves the Order and changes its items through domain methods rather than directly updating individual order-item rows from unrelated services.

The repository then persists the complete relevant state according to the aggregate's consistency contract.

This does not require loading every related object in a large business domain. Aggregate boundaries should remain small enough to preserve meaningful invariants without creating unnecessarily large transactions.

For more about aggregate roots and domain ownership, read Defining Service Boundaries with Domain-Driven Design.

Queries, Filtering, and Pagination

A repository should expose useful queries without allowing arbitrary storage concerns to spread into application services.

For example:

from dataclasses import dataclass
from datetime import datetime
from typing import Protocol

from app.domain import Order


@dataclass(frozen=True)
class OrderPage:
    orders: tuple[Order, ...]
    next_cursor: str | None


class OrderRepository(Protocol):
    def find_pending_before(
        self,
        cutoff: datetime,
        limit: int,
        cursor: str | None = None,
    ) -> OrderPage:
        ...

A domain-oriented query can include pagination without exposing SQLAlchemy query objects or database sessions.

For large datasets, avoid returning every aggregate in memory. Use appropriate indexes, bounded page sizes, and stable ordering. Keyset pagination can be preferable to deep offset pagination for frequently accessed large tables.

Not every read query needs to reconstruct a full domain aggregate. Reporting screens and analytical queries can use dedicated read models or query services when their requirements differ from domain behavior.

Repository vs. Unit of Work

Repository and Unit of Work solve related but different problems.

The repository abstracts access to domain objects. Unit of Work coordinates the transaction across one or more persistence operations.

Suppose one use case must update an order, reserve local inventory records, and insert an outbox message atomically.

If three repositories commit independently, a failure after the first commit can leave the operation partially applied.

An explicit Unit of Work can provide one transaction boundary covering all required database changes.

For a small use case performing a single atomic write, a repository-owned transaction can be reasonable. For multi-repository workflows, transaction coordination should be explicit at the use-case level through a suitable abstraction.

This distinction will be explored in the next article about the Unit of Work Pattern.

Production Design Example: Order Management

Consider an e-commerce application that stores orders in PostgreSQL. The application supports creating an order, retrieving it, and marking it as paid after an authorized payment.

Order creation and status updates must remain correct when multiple requests arrive simultaneously.

Business and Technical Requirements

The requirements are:

  • Order IDs must be unique.
  • Each order belongs to a customer and has a positive monetary amount.
  • New orders begin in the pending state.
  • Only pending orders can transition to paid.
  • Concurrent updates must not silently overwrite one another.
  • Business rules must be testable without PostgreSQL.
  • Database-specific types must not escape into application use cases.

The Order entity enforces valid state transitions. The repository handles persistence and optimistic concurrency. The application service coordinates the operation, and the HTTP adapter translates outcomes into status codes.

Atomic Writes and Optimistic Concurrency

Two common concurrency problems need separate solutions.

The first is duplicate creation. A read-before-write check is not safe because two requests can both observe that the order does not exist. Use a database uniqueness constraint with an atomic insertion strategy.

The second is the lost update. Two clients may read the same version of an order, make different changes, and attempt to save them.

Optimistic concurrency control adds a version number. An update succeeds only if the stored version still matches the version originally read by the application.

For example, an order at version 1 is updated using the condition version = 1. On success, its version becomes 2. A second update based on version 1 fails instead of silently overwriting newer data.

That behavior belongs in the repository's persistence contract, while the domain model determines whether the requested status change is allowed.

Ready-to-Use Example with FastAPI and PostgreSQL

This implementation provides an Order domain entity, an application-owned repository interface, a SQLAlchemy/PostgreSQL implementation, and FastAPI endpoints.

It also includes an in-memory repository with equivalent version-check behavior for isolated unit tests. The example is intentionally smaller than a complete checkout system so the persistence boundary remains easy to inspect.

Project Structure and Dependencies

Create the following project structure, including empty __init__.py files in Python packages:

repository-orders/
  app/
    __init__.py
    domain.py
    use_cases.py
    postgres.py
    api.py
  tests/
    test_orders.py
  requirements.txt
  docker-compose.yml

Create requirements.txt:

fastapi>=0.115,<1
uvicorn>=0.30,<1
sqlalchemy>=2.0,<3
psycopg[binary]>=3.2,<4
pytest>=8,<10
httpx>=0.27,<1

These version ranges support an educational example. Use a tested lockfile and reproducible builds for production deployments.

Domain Model and Repository Contract

The domain entity owns business invariants and state transitions. The repository contract defines the operations application services require without exposing database models.

Create app/domain.py and app/use_cases.py:

# app/domain.py

from dataclasses import dataclass, replace


@dataclass(frozen=True)
class Order:
    order_id: str
    customer_id: str
    amount_cents: int
    status: str = "pending"
    version: int = 1

    def __post_init__(self) -> None:
        if not self.order_id.strip():
            raise ValueError("Order ID is required")

        if not self.customer_id.strip():
            raise ValueError("Customer ID is required")

        if (
            type(self.amount_cents) is not int
            or self.amount_cents <= 0
        ):
            raise ValueError("Amount must be positive")

        if self.status not in {"pending", "paid"}:
            raise ValueError("Invalid order status")

        if self.version < 1:
            raise ValueError("Invalid version")

    def mark_paid(self) -> "Order":
        if self.status != "pending":
            raise ValueError(
                "Only pending orders can be paid"
            )

        # A new domain state, not yet persisted.
        return replace(self, status="paid")


# app/use_cases.py

from typing import Protocol

from app.domain import Order


class OrderAlreadyExists(Exception):
    pass


class OrderNotFound(Exception):
    pass


class OrderConflict(Exception):
    pass


class OrderRepository(Protocol):
    def get_by_id(
        self,
        order_id: str,
    ) -> Order | None:
        ...

    def add(self, order: Order) -> bool:
        """False means the ID exists."""

    def save(
        self,
        order: Order,
        expected_version: int,
    ) -> bool:
        """False means the version no longer matches."""


class CreateOrder:
    def __init__(
        self,
        repository: OrderRepository,
    ) -> None:
        self.repository = repository

    def execute(
        self,
        order_id: str,
        customer_id: str,
        amount_cents: int,
    ) -> Order:
        order = Order(
            order_id=order_id,
            customer_id=customer_id,
            amount_cents=amount_cents,
        )

        if not self.repository.add(order):
            raise OrderAlreadyExists(order_id)

        return order


class MarkOrderPaid:
    def __init__(
        self,
        repository: OrderRepository,
    ) -> None:
        self.repository = repository

    def execute(
        self,
        order_id: str,
    ) -> Order:
        existing = self.repository.get_by_id(
            order_id
        )

        if existing is None:
            raise OrderNotFound(order_id)

        updated = existing.mark_paid()

        if not self.repository.save(
            updated,
            expected_version=existing.version,
        ):
            raise OrderConflict(
                "The order was modified concurrently"
            )

        # The successful write increments the version.
        return replace(
            updated,
            version=existing.version + 1,
        )


class GetOrder:
    def __init__(
        self,
        repository: OrderRepository,
    ) -> None:
        self.repository = repository

    def execute(self, order_id: str) -> Order:
        order = self.repository.get_by_id(order_id)

        if order is None:
            raise OrderNotFound(order_id)

        return order

The domain is immutable, and its mark_paid() method returns a new object representing the requested state. It cannot change a paid order back into pending or mark an already paid order as paid again.

The application layer knows the expected-version contract, but it does not know how SQLAlchemy executes conditional updates.

The example assumes save() is called with an order already retrieved by the use case. More complex systems may return a structured conflict result to distinguish missing aggregates, stale versions, and other persistence failures.

Application Use Cases

The three use cases in app/use_cases.py represent application behavior: creating, retrieving, and marking orders as paid.

Notice that they use a single repository contract rather than exposing SQL queries or opening sessions themselves.

This separation also makes it possible to introduce another driving adapter, such as a command-line tool or message consumer, without rewriting domain validation.

For real payment processing, the MarkOrderPaid operation should be invoked only after trustworthy payment authorization or capture confirmation. This demonstration does not integrate a payment provider and should not be used as a production payment endpoint without authentication, authorization, and payment-state verification.

PostgreSQL Repository with SQLAlchemy

The concrete repository maps between the Order domain entity and the PostgreSQL table.

It implements duplicate-safe inserts and version-checked updates using atomic database operations.

Create app/postgres.py:

import os

from sqlalchemy import (
    Integer,
    String,
    create_engine,
    update,
)
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
)

from app.domain import Order


DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg://app:app@localhost:5432/orders",
)

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
)


class Base(DeclarativeBase):
    pass


class OrderRow(Base):
    __tablename__ = "orders"

    order_id: Mapped[str] = mapped_column(
        String(100),
        primary_key=True,
    )

    customer_id: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )

    amount_cents: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )

    status: Mapped[str] = mapped_column(
        String(30),
        nullable=False,
    )

    version: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
        default=1,
    )


class PostgresOrderRepository:
    def __init__(self, session: Session):
        self.session = session

    def get_by_id(
        self,
        order_id: str,
    ) -> Order | None:
        row = self.session.get(
            OrderRow,
            order_id,
        )

        if row is None:
            return None

        return Order(
            order_id=row.order_id,
            customer_id=row.customer_id,
            amount_cents=row.amount_cents,
            status=row.status,
            version=row.version,
        )

    def add(self, order: Order) -> bool:
        statement = (
            insert(OrderRow)
            .values(
                order_id=order.order_id,
                customer_id=order.customer_id,
                amount_cents=order.amount_cents,
                status=order.status,
                version=order.version,
            )
            .on_conflict_do_nothing(
                index_elements=["order_id"]
            )
            .returning(OrderRow.order_id)
        )

        inserted = self.session.execute(
            statement
        ).scalar_one_or_none()

        if inserted is None:
            self.session.rollback()
            return False

        self.session.commit()
        return True

    def save(
        self,
        order: Order,
        expected_version: int,
    ) -> bool:
        statement = (
            update(OrderRow)
            .where(
                OrderRow.order_id == order.order_id,
                OrderRow.version == expected_version,
            )
            .values(
                status=order.status,
                version=expected_version + 1,
            )
            .returning(OrderRow.order_id)
        )

        updated = self.session.execute(
            statement
        ).scalar_one_or_none()

        if updated is None:
            self.session.rollback()
            return False

        self.session.commit()
        return True

PostgreSQL's primary key ensures only one request can create a given order ID. The conditional update ensures a stale version cannot overwrite the latest committed version.

Each write commits independently in this intentionally small example. The repository owns the transaction because each use case performs one atomic persistence operation.

For a use case that updates multiple aggregates and outbox records together, this transaction ownership would be insufficient. A shared Unit of Work should control the overall commit.

Also note that get_by_id() returns a plain domain object. The ORM model remains inside the persistence adapter.

FastAPI Adapter and Dependency Injection

The HTTP adapter supplies a concrete repository and translates application outcomes into HTTP responses.

Create app/api.py:

from contextlib import asynccontextmanager
from typing import Annotated, Iterator

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.postgres import (
    Base,
    PostgresOrderRepository,
    engine,
)
from app.use_cases import (
    CreateOrder,
    GetOrder,
    MarkOrderPaid,
    OrderAlreadyExists,
    OrderConflict,
    OrderNotFound,
)


class CreateOrderRequest(BaseModel):
    order_id: str = Field(min_length=1)
    customer_id: str = Field(min_length=1)
    amount_cents: int = Field(gt=0)


class OrderResponse(BaseModel):
    order_id: str
    customer_id: str
    amount_cents: int
    status: str
    version: int


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Demo only: production uses migrations.
    Base.metadata.create_all(bind=engine)
    yield
    engine.dispose()


app = FastAPI(
    title="Repository Pattern Orders API",
    lifespan=lifespan,
)


def get_session() -> Iterator[Session]:
    with Session(engine) as session:
        yield session


def get_repository(
    session: Annotated[
        Session,
        Depends(get_session),
    ],
) -> PostgresOrderRepository:
    return PostgresOrderRepository(session)


@app.post(
    "/orders",
    response_model=OrderResponse,
    status_code=201,
)
def create_order(
    request: CreateOrderRequest,
    repository: Annotated[
        PostgresOrderRepository,
        Depends(get_repository),
    ],
):
    try:
        return CreateOrder(repository).execute(
            order_id=request.order_id,
            customer_id=request.customer_id,
            amount_cents=request.amount_cents,
        )

    except OrderAlreadyExists as exc:
        raise HTTPException(
            status_code=409,
            detail="Order already exists",
        ) from exc


@app.get(
    "/orders/{order_id}",
    response_model=OrderResponse,
)
def get_order(
    order_id: str,
    repository: Annotated[
        PostgresOrderRepository,
        Depends(get_repository),
    ],
):
    try:
        return GetOrder(repository).execute(
            order_id
        )

    except OrderNotFound as exc:
        raise HTTPException(
            status_code=404,
            detail="Order not found",
        ) from exc


@app.post(
    "/orders/{order_id}/mark-paid",
    response_model=OrderResponse,
)
def mark_paid(
    order_id: str,
    repository: Annotated[
        PostgresOrderRepository,
        Depends(get_repository),
    ],
):
    try:
        return MarkOrderPaid(repository).execute(
            order_id
        )

    except OrderNotFound as exc:
        raise HTTPException(
            status_code=404,
            detail="Order not found",
        ) from exc

    except OrderConflict as exc:
        raise HTTPException(
            status_code=409,
            detail=str(exc),
        ) from exc

    except ValueError as exc:
        raise HTTPException(
            status_code=409,
            detail=str(exc),
        ) from exc

FastAPI is responsible for HTTP input validation, routing, dependency creation, and response serialization.

The business rules remain in the domain and use-case layers. Application exceptions are mapped to HTTP status codes only at the external boundary.

The endpoints use synchronous SQLAlchemy sessions and synchronous route functions. An asynchronous version would require compatible async database sessions and application interfaces.

Unit Tests with an In-Memory Repository

A repository abstraction becomes particularly useful when application behavior can be tested with a lightweight fake.

The fake must preserve the same important rules as the production implementation, including uniqueness and version checks.

Create tests/test_orders.py:

from dataclasses import replace

import pytest

from app.domain import Order
from app.use_cases import (
    CreateOrder,
    GetOrder,
    MarkOrderPaid,
    OrderAlreadyExists,
    OrderConflict,
)


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

    def get_by_id(
        self,
        order_id: str,
    ) -> Order | None:
        return self.orders.get(order_id)

    def add(self, order: Order) -> bool:
        if order.order_id in self.orders:
            return False

        self.orders[order.order_id] = order
        return True

    def save(
        self,
        order: Order,
        expected_version: int,
    ) -> bool:
        existing = self.orders.get(
            order.order_id
        )

        if (
            existing is None
            or existing.version != expected_version
        ):
            return False

        self.orders[order.order_id] = replace(
            order,
            version=expected_version + 1,
        )
        return True


def test_create_and_retrieve_order():
    repository = InMemoryOrderRepository()

    created = CreateOrder(repository).execute(
        order_id="order-1001",
        customer_id="customer-42",
        amount_cents=4999,
    )

    loaded = GetOrder(repository).execute(
        created.order_id
    )

    assert loaded == created
    assert loaded.status == "pending"
    assert loaded.version == 1


def test_duplicate_order():
    repository = InMemoryOrderRepository()
    service = CreateOrder(repository)

    service.execute(
        "order-1001",
        "customer-42",
        4999,
    )

    with pytest.raises(OrderAlreadyExists):
        service.execute(
            "order-1001",
            "customer-42",
            4999,
        )


def test_mark_order_paid():
    repository = InMemoryOrderRepository()

    CreateOrder(repository).execute(
        "order-1001",
        "customer-42",
        4999,
    )

    result = MarkOrderPaid(repository).execute(
        "order-1001"
    )

    assert result.status == "paid"
    assert result.version == 2


def test_stale_version_rejected():
    repository = InMemoryOrderRepository()

    order = Order(
        order_id="order-1001",
        customer_id="customer-42",
        amount_cents=4999,
    )

    assert repository.add(order)

    # A concurrent writer commits version 2.
    assert repository.save(
        order.mark_paid(),
        expected_version=1,
    )

    # A stale update based on version 1 fails.
    assert not repository.save(
        order,
        expected_version=1,
    )


def test_application_handles_conflict():
    repository = InMemoryOrderRepository()

    CreateOrder(repository).execute(
        "order-1001",
        "customer-42",
        4999,
    )

    # Simulate a concurrency conflict from storage.
    class ConflictingRepository(
        InMemoryOrderRepository
    ):
        def save(self, order, expected_version):
            return False

    conflicting = ConflictingRepository()
    conflicting.orders = dict(repository.orders)

    with pytest.raises(OrderConflict):
        MarkOrderPaid(conflicting).execute(
            "order-1001"
        )

These tests exercise the real domain and use-case code without opening a database connection.

They verify application behavior, not PostgreSQL correctness. Integration tests must also confirm database constraints, version updates, transaction behavior, and mapping logic under actual concurrent requests.

Run and Verify the Application

Use Docker Compose for a local PostgreSQL instance.

Create docker-compose.yml:

services:
  postgres:
    image: postgres:16-alpine

    environment:
      POSTGRES_DB: orders
      POSTGRES_USER: app
      POSTGRES_PASSWORD: app

    ports:
      - "127.0.0.1:5432:5432"

    healthcheck:
      test:
        - CMD-SHELL
        - pg_isready -U app -d orders
      interval: 5s
      timeout: 3s
      retries: 10

Run the application and verify the repository behavior:

# Start PostgreSQL.
docker compose up -d --wait

# Create and activate a Python environment.
python3 -m venv .venv
source .venv/bin/activate

# Install dependencies.
pip install -r requirements.txt

# Run unit tests without PostgreSQL.
pytest -q

# Configure the database connection.
export DATABASE_URL="postgresql+psycopg://app:app@localhost:5432/orders"

# Start the API.
uvicorn app.api:app --reload

# In another terminal, create an order.
curl -i -X POST \
  http://127.0.0.1:8000/orders \
  -H "Content-Type: application/json" \
  -d '{
    "order_id": "order-1001",
    "customer_id": "customer-42",
    "amount_cents": 4999
  }'

# Retrieve the order.
curl -i \
  http://127.0.0.1:8000/orders/order-1001

# Mark the order as paid.
curl -i -X POST \
  http://127.0.0.1:8000/orders/order-1001/mark-paid

# Retrieve the updated version.
curl -i \
  http://127.0.0.1:8000/orders/order-1001

# Remove local infrastructure when finished.
docker compose down

The first request should return an order with status pending and version 1. After marking it paid, the order should have status paid and version 2.

Repeated creation requests return 409 Conflict. Attempting to mark an already paid order as paid also returns a conflict.

The schema is created automatically for this local demonstration. A production application should use versioned migrations, managed database credentials, and connection-pool configuration appropriate to its deployment environment.

When to Use the Repository Pattern

The Repository Pattern is useful when it establishes a meaningful boundary between business logic and persistence. It is less valuable when it simply duplicates an ORM's existing interface.

Where Repositories Provide Value

Repositories are especially useful when:

  • Domain models contain meaningful business rules and invariants.
  • Several use cases depend on the same persistence behavior.
  • Business services must remain independent of ORM models.
  • Database queries require reusable domain-oriented logic.
  • Application tests benefit from lightweight fake persistence.
  • Aggregate boundaries require explicit consistency contracts.
  • Infrastructure implementations may evolve independently of the application core.

They fit naturally into Hexagonal, Onion, and Clean Architecture because those designs emphasize dependency boundaries.

When a Repository Adds Unnecessary Complexity

A repository may be unnecessary for a simple CRUD application where the ORM already provides a clear, well-contained data access API.

For example, a small administrative endpoint that reads one configuration table may not justify a repository interface, concrete repository class, mapping layer, and separate service if those components only forward calls.

Repositories can also be less appropriate for read-heavy analytical endpoints that return projections rather than domain aggregates.

In those cases, a dedicated query module or read-model service may be clearer and more efficient than forcing the query through an aggregate repository.

The important question is whether the abstraction protects meaningful application behavior, not whether every database table has a matching repository class.

Common Mistakes and Production Practices

A repository can make application code cleaner, but it cannot eliminate database constraints, transaction design, or query-performance considerations.

Its implementation must preserve the semantics promised by the interface.

Common Repository Pattern Mistakes

Common problems include:

  • Generic CRUD interfaces everywhere: The repository provides no meaningful domain abstraction.
  • One repository per database table: Ignores aggregate boundaries and business responsibilities.
  • Leaking ORM models: Application services remain coupled to persistence.
  • Exposing arbitrary query objects: SQLAlchemy or database APIs escape through the contract.
  • Using read-before-write uniqueness checks: Concurrent inserts can violate assumptions.
  • Ignoring optimistic locking: Concurrent updates silently overwrite newer state.
  • Committing each repository independently: Multi-repository business operations lose atomicity.
  • Hiding expensive queries: N+1 behavior or unbounded reads become difficult to recognize.
  • Assuming fake repositories prove database correctness: Transaction and isolation behavior remains untested.
  • Assuming database replacement is effortless: Query capabilities and consistency semantics differ.
  • Swallowing infrastructure errors: Temporary outages are misreported as missing records.
  • Introducing unnecessary abstractions: Simple code becomes harder to follow.

In particular, a repository should not convert every database error into None or False. A connection failure is not the same as a missing order, and a deadlock is not necessarily a domain conflict.

Production Checklist

Before using repositories in production, verify that:

  • Repository methods reflect application and domain requirements.
  • Domain objects remain independent of database libraries.
  • Persistence mappings do not leak into use cases.
  • Critical uniqueness rules are enforced by the database.
  • Concurrent updates use appropriate locking or version checks.
  • Transaction ownership is explicitly defined.
  • Multi-repository operations share a transaction where required.
  • Queries are bounded and use appropriate indexes.
  • Pagination uses stable ordering.
  • Database failures are distinguished from business conflicts.
  • Connections and sessions have controlled lifetimes.
  • Schema changes use versioned migrations.
  • Unit tests cover use cases with test repositories.
  • Integration tests validate real SQL and transaction behavior.
  • Operational metrics expose slow queries, contention, and connection exhaustion.

Frequently Asked Questions

The Repository Pattern is frequently discussed alongside ORMs, DAOs, Domain-Driven Design, and Unit of Work. These concepts are related but do not always require separate classes.

Is SQLAlchemy Already a Repository?

SQLAlchemy provides powerful ORM mapping, querying, sessions, and transaction management. Some of its capabilities overlap with repositories.

A separate repository is useful when it adds a domain-oriented interface and protects application logic from SQLAlchemy-specific details. Otherwise, it may only duplicate existing functionality.

Do You Need a Repository for Every Table?

No. In DDD, repositories commonly correspond to aggregate roots rather than individual database tables.

One aggregate may be persisted across multiple tables, while some tables may not need a repository abstraction at all.

Does Every Repository Need an Interface?

No. An interface is valuable when it establishes a meaningful dependency boundary or allows independent implementation and testing.

A small application may use a concrete repository directly without sacrificing maintainability.

Should a Repository Commit Transactions?

It depends on the application's transaction model.

For one atomic persistence operation, repository-owned commits may be sufficient. When a use case coordinates multiple repositories, an application-level Unit of Work often provides clearer transaction ownership.

Does the Repository Pattern Affect Performance?

The method-call abstraction usually contributes negligible overhead compared with database and network operations.

However, poor repository design can hide inefficient queries, repeated database access, excessive object mapping, or N+1 query behavior.

Use query profiling and integration tests to validate performance rather than assuming the abstraction itself is free or expensive.

Conclusion

The Repository Pattern provides a domain-oriented boundary between application logic and data persistence. By encapsulating database access and mapping, it allows application services to operate on business objects without knowing the details of SQL queries, ORM models, and database sessions.

Its greatest value appears in applications with meaningful domain behavior, multiple persistence operations, or important testing requirements. Combined with dependency inversion, it fits naturally into Hexagonal, Onion, and Clean Architecture.

However, repository abstractions should not be introduced mechanically. A generic CRUD wrapper around an ORM may add more complexity than value, while a repository that hides transaction or query behavior can make production problems harder to diagnose.

Key Takeaway: Design repositories around domain operations and consistency guarantees, not database tables. Keep transaction boundaries explicit, enforce concurrency rules atomically, and test both the application contract and the real persistence implementation.

Continue exploring related patterns and implementation techniques:

For the original pattern description, see Martin Fowler's Repository Pattern.

Comments (0)

Related articles