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 transaction

These 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 Order is created
Each Product's stock is decremented
A Payment record 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)

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

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

# 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)
        )

🔄 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()

🚀 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

✅ 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:")
        )

🗂️ 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

🔑 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*

Enterprise Design Patterns in Python: Repository & Unit of Work — Real-World E-Commerce Example

Enterprise Design Patterns in Python: Repository & Unit of Work 🐍🏗️

🧠 What Are Enterprise Design Patterns?

🛒 The Problem: An E-Commerce Order System

📁 Repository Pattern

Definition

Domain Model

Abstract Repository Interface

Concrete Implementations

🔄 Unit of Work Pattern

Definition

Implementation

🚀 Service Layer: Putting It All Together

✅ Testing: The Real Payoff

🗂️ Project Structure

🔑 Key Takeaways

📚 References & Further Reading

💬 Comment / Peer Review Section

Tags

Author

Stats

Published

You Might Also Like

Some friends wanted to see how I use DigitalOcean. So I built them the smallest real app I could.

The LLM Visibility Tools Cost $79/Month. Mine is Open Source.

On programming languages, targets, and platforms

My eval harness paid for itself on the first run: 0.57 0.96, two bugs no unit test could catch

Never lose a training run again: a checkpoint-and-resume playbook for ephemeral GPUs

I almost added an em-dash remover to my LLM library. Then I tested whether local models even produce em-dashes.