Keeping the Domain Pure
One of DDD's core goals is keeping your domain model free of infrastructure concerns. Your Order class should contain business logic — not SQL queries, not HTTP calls, not Redis commands. If your domain objects are littered with database calls, they become impossible to test in isolation and tightly coupled to your infrastructure choices.
The Repository Pattern
A Repository is an abstraction that makes your collection of aggregate roots look like an in-memory collection, hiding all the persistence details behind a clean interface.
from abc import ABC, abstractmethod
# The interface — defined in the domain layer
class OrderRepository(ABC):
@abstractmethod
def find_by_id(self, order_id: str) -> Order | None:
pass
@abstractmethod
def save(self, order: Order) -> None:
pass
@abstractmethod
def find_by_customer(self, customer_id: str) -> list[Order]:
pass
# The implementation — defined in the infrastructure layer
class PostgresOrderRepository(OrderRepository):
def __init__(self, db_connection):
self._db = db_connection
def find_by_id(self, order_id: str) -> Order | None:
row = self._db.execute(
"SELECT * FROM orders WHERE order_id = %s", [order_id]
).fetchone()
if not row:
return None
return self._reconstruct_order(row)
def save(self, order: Order) -> None:
# Upsert the order and all its items atomically
self._db.execute(
"INSERT INTO orders ... ON CONFLICT DO UPDATE ...",
[order.order_id, order.customer_id, order.status]
)
for item in order._items:
self._db.execute("INSERT INTO order_items ...", [...])
The domain layer depends on the OrderRepository interface. The infrastructure layer provides the PostgresOrderRepository implementation. This means you can swap PostgreSQL for DynamoDB or an in-memory store for tests without changing a single line of domain logic.
One repository per aggregate root is the rule. You don't have a repository for OrderItem — you always access items through the Order repository.