Research Group Activity 02 — Enterprise Design Patterns
Author: Sergio Colque Ponce · Code: github.com/srg-cp/enterprise-repository-pattern
Every non-trivial application eventually faces the same question: where does the data-access code live? Sprinkle SQL across your business logic and you get a codebase that is impossible to test and painful to change. Martin Fowler's Patterns of Enterprise Application Architecture (PoEAA) catalogs battle-tested answers to exactly these problems.
In this article I focus on the Repository pattern, and combine it with two of its natural companions from the same catalog — Unit of Work and Service Layer — using a small, runnable Python example.
What is the Repository pattern?
Fowler defines it as a pattern that mediates between the domain and data mapping layers using a collection-like interface for accessing domain objects.
In plain words: your business code talks to something that looks like an in-memory collection — add, get, list, remove — and has no idea whether the data actually lives in PostgreSQL, SQLite, a REST API, or a Python dictionary. The repository is the wall between "what my app does" and "how data is stored".
The problem it solves
Without it, data access leaks everywhere:
# The anti-pattern: SQL glued to business logic
def sell_product(sku, qty):
conn = sqlite3.connect("catalog.db")
row = conn.execute("SELECT stock FROM products WHERE sku=?", (sku,)).fetchone()
if row[0] < qty:
raise Exception("no stock")
conn.execute("UPDATE products SET stock = stock - ? WHERE sku=?", (qty, sku))
conn.commit()
This function is impossible to unit-test without a database, mixes three concerns (persistence, transactions, business rules), and hard-codes SQLite forever. The Repository pattern untangles all of that.
The example: a product catalog
Our domain is a tiny inventory system. The domain object knows its business rules and nothing about storage:
from dataclasses import dataclass
from decimal import Decimal
class InsufficientStock(Exception):
...
@dataclass
class Product:
sku: str
name: str
price: Decimal
stock: int = 0
def sell(self, quantity: int) -> None:
if quantity > self.stock:
raise InsufficientStock(
f"cannot sell {quantity} units of {self.sku}; only {self.stock} in stock"
)
self.stock -= quantity
Step 1 — Define the interface
The whole application will depend on this abstraction, never on a concrete database:
from abc import ABC, abstractmethod
class ProductRepository(ABC):
@abstractmethod
def add(self, product: Product) -> None: ...
@abstractmethod
def get(self, sku: str) -> Product | None: ...
@abstractmethod
def list(self) -> list[Product]: ...
@abstractmethod
def update(self, product: Product) -> None: ...
@abstractmethod
def remove(self, sku: str) -> None: ...
Step 2 — A real implementation (SQLite)
This concrete repository also plays the role of a small Data Mapper: it translates between the plain Product object and its relational rows. Notice prices are stored as integer cents to dodge floating-point errors — a classic enterprise concern.
class SqliteProductRepository(ProductRepository):
def __init__(self, connection):
self._conn = connection
self._conn.row_factory = sqlite3.Row
self._conn.execute(
"""CREATE TABLE IF NOT EXISTS products (
sku TEXT PRIMARY KEY, name TEXT NOT NULL,
price_cents INTEGER NOT NULL, stock INTEGER NOT NULL)"""
)
def get(self, sku):
row = self._conn.execute(
"SELECT * FROM products WHERE sku = ?", (sku,)
).fetchone()
if row is None:
return None
return Product(row["sku"], row["name"],
Decimal(row["price_cents"]) / 100, row["stock"])
# add / list / update / remove follow the same idea...
Step 3 — A fake implementation for tests
Here is the payoff. Because it obeys the same contract, this in-memory version is a drop-in replacement — no database needed, and it runs in microseconds:
import copy
class InMemoryProductRepository(ProductRepository):
def __init__(self):
self._items: dict[str, Product] = {}
def add(self, product):
# store a copy so external mutations don't leak into the repo
self._items[product.sku] = copy.deepcopy(product)
def get(self, sku):
found = self._items.get(sku)
return copy.deepcopy(found) if found else None
# ...
Step 4 — Service Layer: business code that ignores storage
The Service Layer holds the use cases. It receives a ProductRepository by constructor (dependency injection) and depends only on the interface:
class CatalogService:
def __init__(self, products: ProductRepository):
self._products = products
def sell(self, sku: str, quantity: int) -> Product:
product = self._products.get(sku)
if product is None:
raise ValueError(f"product {sku} not found")
product.sell(quantity) # business rule
self._products.update(product) # persistence, abstracted away
return product
Read that sell method again: there is not a single line of SQL. It works identically with SQLite in production and with the in-memory fake in tests.
Step 5 — Unit of Work: atomic transactions
A single business operation often touches several objects and must be all-or-nothing. That is the Unit of Work pattern. Here it wraps the connection and, used as a context manager, guarantees commit/rollback and cleanup:
class UnitOfWork:
def __enter__(self):
self._conn = sqlite3.connect(self._database)
self.products = SqliteProductRepository(self._conn)
return self
def __exit__(self, exc_type, exc, tb):
if exc_type is not None:
self.rollback() # any error → undo everything
self.close()
return False
def commit(self): self._conn.commit()
def rollback(self): self._conn.rollback()
Usage:
with UnitOfWork("catalog.db") as uow:
uow.products.add(Product("BOOK-1", "PoEAA", Decimal("54.99"), 5))
uow.commit() # nothing persists unless we reach this line
Testing: one suite, two backends
Because both repositories share the same interface, a single test suite can run against both via a parametrized fixture — proving they are truly interchangeable:
@pytest.fixture(params=["memory", "sqlite"])
def repository(request):
if request.param == "memory":
yield InMemoryProductRepository()
else:
conn = sqlite3.connect(":memory:")
yield SqliteProductRepository(conn)
conn.close()
def test_update_persists_changes(repository):
product = Product("SKU-1", "Widget", Decimal("9.99"), 10)
repository.add(product)
product.sell(3)
repository.update(product)
assert repository.get("SKU-1").stock == 7
Running pytest executes each contract test twice — once per backend — for 19 green tests total.
Automation (CI)
The repo ships with a GitHub Actions workflow that lints with ruff and runs pytest on Python 3.9, 3.10, 3.11 and 3.12 on every push and pull request:
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.9", "3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with: { python-version: "${{ matrix.python-version }}" }
- run: pip install -e ".[dev]"
- run: ruff check .
- run: pytest
Why these patterns pay off
- Testable — swap the database for an in-memory fake; no I/O in unit tests.
- Flexible — migrating from SQLite to Postgres touches one class, not your business logic.
-
Readable — use cases express intent (
sell,restock), notINSERT/UPDATE. - Safe — Unit of Work keeps multi-step operations atomic.
The trade-off is extra indirection, so for a throwaway script it may be overkill. But for any application meant to live and grow — exactly what "enterprise" means — this structure repays its cost many times over.
Conclusion
The Repository pattern draws a clean line between your domain and your database. Paired with a Unit of Work for transactions and a Service Layer for use cases, you get code that is easy to test, easy to change, and easy to read — the enduring lesson of Fowler's PoEAA catalog.
Full, runnable source (with CI): github.com/srg-cp/enterprise-repository-pattern
Written for Research Group Activity 02 — Enterprise Design Patterns.
Top comments (0)