Enterprise Design Patterns in Python: Repository & Unit of Work 🐍🏗️
Series: Enterprise Application Architecture | Source: Fowler’s EAA Catalog | Code: GitHub Repository
🧠 What Are Enterprise Design Patterns?
Martin Fowler’s Patterns of Enterprise Application Architecture (2002) is one of the most influential books in software engineering. It documents recurring architectural solutions — patterns — that solve common problems in enterprise systems: how to organize domain logic, how to talk to databases, how to handle transactions, and more.
In this article, we’ll explore two of the most powerful and widely-used patterns from that catalog:
Pattern Category Core Purpose Repository Data Source Abstracts data access behind a collection-like interface Unit of Work Data Source Tracks object changes and commits them as a single transactionThese two patterns work beautifully together — and you’ll see exactly why with a real-world example.
🛒 The Problem: An E-Commerce Order System
Imagine you’re building a backend for an online store. When a customer places an order:
- A new
Orderis created - Each
Product‘s stock is decremented - A
Paymentrecord is registered
If any of these steps fail midway, the entire operation should roll back — no partial state. This is exactly the problem the Unit of Work pattern solves, and the Repository pattern makes it all cleanly testable.
📁 Repository Pattern
Definition
“A Repository mediates between the domain and data mapping layers using a collection-like interface for accessing domain objects.”
— Martin Fowler, PoEAA
The Repository acts as an in-memory collection of domain objects. Your business logic never knows if it’s talking to PostgreSQL, SQLite, or even a mock list — it just calls .add(), .get(), .list().
Domain Model
# models.py
from dataclasses import dataclass, field
from typing import List
from uuid import uuid4
@dataclass
class Product:
id: str
name: str
price: float
stock: int
@dataclass
class OrderItem:
product_id: str
quantity: int
unit_price: float
@dataclass
class Order:
id: str = field(default_factory=lambda: str(uuid4()))
customer_id: str = ""
items: List[OrderItem] = field(default_factory=list)
status: str = "pending"
def add_item(self, product: Product, quantity: int):
if product.stock < quantity:
raise ValueError(f"Insufficient stock for {product.name}")
self.items.append(OrderItem(
product_id=product.id,
quantity=quantity,
unit_price=product.price
))
@property
def total(self) -> float:
return sum(item.quantity * item.unit_price for item in self.items)
Enter fullscreen mode Exit fullscreen mode
Abstract Repository Interface
# repositories/base.py
from abc import ABC, abstractmethod
from typing import Generic, List, Optional, TypeVar
T = TypeVar("T")
class AbstractRepository(ABC, Generic[T]):
@abstractmethod
def add(self, entity: T) -> None:
raise NotImplementedError
@abstractmethod
def get(self, entity_id: str) -> Optional[T]:
raise NotImplementedError
@abstractmethod
def list(self) -> List[T]:
raise NotImplementedError
Enter fullscreen mode Exit fullscreen mode
Concrete Implementations
# repositories/order_repository.py
import sqlite3
import json
from typing import List, Optional
from models import Order, OrderItem
from repositories.base import AbstractRepository
class SqliteOrderRepository(AbstractRepository[Order]):
def __init__(self, connection: sqlite3.Connection):
self.conn = connection
self._create_table()
def _create_table(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS orders (
id TEXT PRIMARY KEY,
customer_id TEXT NOT NULL,
items TEXT NOT NULL,
status TEXT NOT NULL
)
""")
def add(self, order: Order) -> None:
items_json = json.dumps([
{"product_id": i.product_id, "quantity": i.quantity, "unit_price": i.unit_price}
for i in order.items
])
self.conn.execute(
"INSERT INTO orders (id, customer_id, items, status) VALUES (?, ?, ?, ?)",
(order.id, order.customer_id, items_json, order.status)
)
def get(self, order_id: str) -> Optional[Order]:
cursor = self.conn.execute(
"SELECT id, customer_id, items, status FROM orders WHERE id = ?",
(order_id,)
)
row = cursor.fetchone()
if not row:
return None
items = [OrderItem(**i) for i in json.loads(row[2])]
return Order(id=row[0], customer_id=row[1], items=items, status=row[3])
def list(self) -> List[Order]:
cursor = self.conn.execute("SELECT id, customer_id, items, status FROM orders")
orders = []
for row in cursor.fetchall():
items = [OrderItem(**i) for i in json.loads(row[2])]
orders.append(Order(id=row[0], customer_id=row[1], items=items, status=row[3]))
return orders
Enter fullscreen mode Exit fullscreen mode
# repositories/product_repository.py
import sqlite3
from typing import List, Optional
from models import Product
from repositories.base import AbstractRepository
class SqliteProductRepository(AbstractRepository[Product]):
def __init__(self, connection: sqlite3.Connection):
self.conn = connection
self._create_table()
def _create_table(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS products (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
price REAL NOT NULL,
stock INTEGER NOT NULL
)
""")
def add(self, product: Product) -> None:
self.conn.execute(
"INSERT OR REPLACE INTO products (id, name, price, stock) VALUES (?, ?, ?, ?)",
(product.id, product.name, product.price, product.stock)
)
def get(self, product_id: str) -> Optional[Product]:
cursor = self.conn.execute(
"SELECT id, name, price, stock FROM products WHERE id = ?",
(product_id,)
)
row = cursor.fetchone()
return Product(*row) if row else None
def list(self) -> List[Product]:
cursor = self.conn.execute("SELECT id, name, price, stock FROM products")
return [Product(*row) for row in cursor.fetchall()]
def update_stock(self, product_id: str, new_stock: int) -> None:
self.conn.execute(
"UPDATE products SET stock = ? WHERE id = ?",
(new_stock, product_id)
)
Enter fullscreen mode Exit fullscreen mode
🔄 Unit of Work Pattern
Definition
“A Unit of Work maintains a list of objects affected by a business transaction and coordinates the writing out of changes and the resolution of concurrency problems.”
— Martin Fowler, PoEAA
The UoW ensures that all operations in a business transaction either all succeed or all fail together — like a database transaction, but at the application layer.
Implementation
# unit_of_work.py
import sqlite3
from abc import ABC, abstractmethod
from repositories.order_repository import SqliteOrderRepository
from repositories.product_repository import SqliteProductRepository
class AbstractUnitOfWork(ABC):
orders: SqliteOrderRepository
products: SqliteProductRepository
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type:
self.rollback()
else:
self.commit()
@abstractmethod
def commit(self):
raise NotImplementedError
@abstractmethod
def rollback(self):
raise NotImplementedError
class SqliteUnitOfWork(AbstractUnitOfWork):
def __init__(self, db_path: str = ":memory:"):
self.db_path = db_path
def __enter__(self):
self.conn = sqlite3.connect(self.db_path)
self.conn.execute("PRAGMA journal_mode=WAL")
self.orders = SqliteOrderRepository(self.conn)
self.products = SqliteProductRepository(self.conn)
return super().__enter__()
def commit(self):
self.conn.commit()
def rollback(self):
self.conn.rollback()
def __exit__(self, exc_type, exc_val, exc_tb):
super().__exit__(exc_type, exc_val, exc_tb)
self.conn.close()
Enter fullscreen mode Exit fullscreen mode
🚀 Service Layer: Putting It All Together
# services/order_service.py
from typing import List, Tuple
from models import Order
from unit_of_work import AbstractUnitOfWork
class OrderService:
@staticmethod
def place_order(
customer_id: str,
items: List[Tuple[str, int]], # [(product_id, quantity), ...]
uow: AbstractUnitOfWork
) -> Order:
"""
Places an order atomically. All stock updates and order creation
happen in a single Unit of Work transaction.
"""
with uow:
order = Order(customer_id=customer_id)
for product_id, quantity in items:
product = uow.products.get(product_id)
if not product:
raise ValueError(f"Product {product_id} not found")
# This validates stock internally
order.add_item(product, quantity)
# Decrement stock
product.stock -= quantity
uow.products.update_stock(product.id, product.stock)
order.status = "confirmed"
uow.orders.add(order)
# commit() is called automatically by __exit__
return order
Enter fullscreen mode Exit fullscreen mode
✅ Testing: The Real Payoff
The biggest benefit of these patterns is testability. We can test all business logic without touching a real database:
# tests/test_order_service.py
import pytest
from models import Product
from services.order_service import OrderService
from unit_of_work import SqliteUnitOfWork
DB_PATH = ":memory:"
@pytest.fixture
def uow():
return SqliteUnitOfWork(DB_PATH)
def seed_products(uow: SqliteUnitOfWork):
with uow:
uow.products.add(Product("p1", "Laptop", 999.99, 10))
uow.products.add(Product("p2", "Mouse", 29.99, 50))
def test_place_order_success():
uow = SqliteUnitOfWork(":memory:")
seed_products(uow)
order = OrderService.place_order(
customer_id="customer_001",
items=[("p1", 2), ("p2", 1)],
uow=SqliteUnitOfWork(":memory:") # use fresh UoW in real tests
)
assert order.status == "confirmed"
assert order.total == (2 * 999.99) + 29.99
def test_order_fails_on_insufficient_stock():
uow = SqliteUnitOfWork(":memory:")
with pytest.raises(ValueError, match="Insufficient stock"):
with uow:
uow.products.add(Product("p3", "GPU", 1200.0, 1))
OrderService.place_order(
customer_id="customer_002",
items=[("p3", 5)], # Requesting 5, only 1 in stock
uow=SqliteUnitOfWork(":memory:")
)
Enter fullscreen mode Exit fullscreen mode
🗂️ Project Structure
enterprise-patterns-python/
│
├── models.py # Domain models (Order, Product, OrderItem)
├── unit_of_work.py # UoW abstract + SQLite implementation
├── repositories/
│ ├── __init__.py
│ ├── base.py # AbstractRepository[T]
│ ├── order_repository.py # SqliteOrderRepository
│ └── product_repository.py # SqliteProductRepository
├── services/
│ ├── __init__.py
│ └── order_service.py # OrderService (business logic)
├── tests/
│ └── test_order_service.py # Pytest tests
├── main.py # Demo script
└── requirements.txt
Enter fullscreen mode Exit fullscreen mode
🔑 Key Takeaways
- Repository Pattern decouples your domain logic from data storage — swap SQLite for PostgreSQL or even a mock list without changing a single line of business code.
- Unit of Work Pattern coordinates multiple repository operations as a single atomic transaction — no partial state, no data corruption.
- Together, these patterns produce code that is testable, maintainable, and swappable at the infrastructure layer.
- This approach is the backbone of Clean Architecture and Domain-Driven Design (DDD) in Python projects.
📚 References & Further Reading
- 📖 Martin Fowler — EAA Catalog
- 📖 Repository Pattern — PoEAA
- 📖 Unit of Work — PoEAA
- 📖 Architecture Patterns with Python — cosmic python
- 💻 Full Source Code on GitHub
💬 Comment / Peer Review Section
For teammates reviewing this article: Feel free to write your abstract or observation as a comment below! A great starting point:
“This article demonstrates how the Repository pattern enforces a clean boundary between domain logic and persistence. One important observation is that…”
Written by **Jefferson Rosas Chambilla* — Software Engineering student at Universidad Privada de Tacna*
답글 남기기