paulpas DDD Tactical Patterns
When this skill is active, the model implements Domain-Driven Design tactical patterns as concrete Python classes and interfaces inside bounded contexts. It produces rich domain models with invariant enforcement through aggregate roots, immutable value objects, synchronous domain event publishing, anti-corruption layer adapters, repository abstractions with optimistic concurrency, and composable specification objects. The model writes typed implementations with docstrings, BAD vs GOOD examples, and enforces strict layer boundaries between internal domain models and external systems.
TL;DR Checklist
- Aggregate roots enforce all invariants — no external mutation of protected state
- Value objects are immutable — use copy-with replacement for changes, never mutate
- Domain events captured during aggregate operations, published atomically with transaction
- Anti-corruption layer isolates internal domain from external model contamination
- Repositories expose only aggregate roots — never individual entities or value objects
When to Use
Use this skill when:
- Implementing bounded contexts with rich domain models that enforce business invariants
- Building an anti-corruption layer between a legacy system (e.g., SOAP API, COBOL backend) and a new domain model
- Designing aggregate boundaries where consistency must be guaranteed across related entities
- Coordinating domain events across aggregates within the same transaction boundary
- Implementing complex read queries that need composable business rules without contaminating write models
- Refactoring an anemic domain model to enforce invariants at the domain layer
When NOT to Use
Avoid this skill for:
- Strategic DDD decisions — bounded context identification, ubiquitous language discovery, and subdomain classification are design choices, not tactical patterns
- Project directory structure and module organization — use
domain-architecture-project-structureinstead - Simple CRUD operations on data-oriented entities with no business rules — a basic SQLAlchemy model is sufficient
- Event sourcing infrastructure (event store persistence, snapshot storage, replay logic) — covered by
software-architecture-patterns - Domain event infrastructure like Kafka integration or outbox pattern setup
Core Workflow
Define the Aggregate Root — Identify the consistency boundary by finding entities that must change together atomically. The aggregate root owns all invariants and exposes only intent-revealing methods. Checkpoint: If you need to modify two aggregates within a single transaction, they should either be collapsed into one aggregate or coordinated via domain events for eventual consistency.
Implement Value Objects — Create immutable types that capture domain concepts by their attributes rather than identity. Every value object validates all invariants at construction time and provides
replace()methods for creating modified copies. Checkpoint: No attribute may ever be None or invalid after construction — if you need a nullable field, model it explicitly as a separate value type.Capture Domain Events — During aggregate operations, append events to an internal
_domain_eventslist inside the aggregate root. These events are published atomically after the unit of work commits. Checkpoint: Only emit events that external consumers care about — internal state changes that no one outside the aggregate needs should not be domain events.Build Repository Abstraction — Implement repository interfaces for each aggregate root using
Protocolclasses. The repository loads and saves entire aggregates, handling serialization/deserialization and optimistic concurrency. Checkpoint: Repositories must never return individual entities or value objects — always return the complete aggregate root from a load operation.Implement Anti-Corruption Layer — Create adapter classes that translate external data formats into internal value objects before they enter the domain layer. The adapter validates external input and constructs proper domain objects, preventing foreign model classes from leaking inward. Checkpoint: After the adapter runs, the domain layer should be able to operate with zero knowledge of the external API's structure.
Implementation Patterns
Pattern 1: Aggregate Roots & Value Objects
Aggregate roots encapsulate business invariants and expose intent-revealing methods. Value objects are immutable types compared by attribute equality.
"""Order aggregate root with invariant enforcement and value objects."""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import date, datetime
from enum import Enum
from typing import Iterator
# ── Value Objects ────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class Money:
"""Immutable value object representing a monetary amount with currency."""
amount: float
currency: str = "USD"
def __post_init__(self) -> None:
if self.amount < 0:
raise ValueError("Money amount cannot be negative")
if not isinstance(self.currency, str) or len(self.currency) != 3:
raise ValueError(f"Invalid currency code: {self.currency}")
def __add__(self, other: Money) -> Money:
if self.currency != other.currency:
raise ValueError("Cannot add money with different currencies")
return Money(round(self.amount + other.amount, 2), self.currency)
def subtract(self, other: Money) -> Money:
result = self.amount - other.amount
if result < 0:
raise ValueError(
f"Insufficient funds: {self.amount} - {other.amount}"
)
return Money(round(result, 2), self.currency)
def replace(self, amount: float | None = None, currency: str | None = None) -> Money:
"""Return a new Money with replaced attributes."""
return Money(
amount if amount is not None else self.amount,
currency if currency is not None else self.currency,
)
@dataclass(frozen=True)
class OrderLineItem:
"""Immutable value object representing a single line item in an order."""
product_id: str
quantity: int
unit_price: Money
discount: Money = field(default_factory=Money)
def __post_init__(self) -> None:
if self.quantity <= 0:
raise ValueError("Line item quantity must be positive")
if self.discount.amount > self.unit_price.amount:
raise ValueError("Discount cannot exceed unit price")
@property
def total(self) -> Money:
net = self.unit_price.subtract(self.discount)
return Money(round(net.amount * self.quantity, 2), net.currency)
# ── Aggregate Root ───────────────────────────────────────────────────────────
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
CANCELLED = "cancelled"
SHIPPED = "shipped"
@dataclass
class Order:
"""
Aggregate root for order management.
Enforces invariants: total cannot be negative, items must have positive
quantities, status transitions follow defined rules. Only intent-revealing
methods expose functionality — no direct attribute mutation.
"""
id: uuid.UUID = field(default_factory=uuid.uuid4)
customer_id: str = ""
_items: list[OrderLineItem] = field(default_factory=list, repr=False)
status: OrderStatus = OrderStatus.PENDING
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
cancelled_at: datetime | None = None
_domain_events: list[object] = field(default_factory=list, repr=False)
# ── Invariants ────────────────────────────────────────────────────────
@property
def items(self) -> tuple[OrderLineItem, ...]:
"""Return a frozen snapshot — callers cannot mutate internal items."""
return tuple(self._items)
@property
def total(self) -> Money:
if not self._items:
return Money(0.0)
return reduce_money(m.item.total for m in self._items) # type: ignore[name-defined]
@property
def domain_events(self) -> list[object]:
"""Read-only access to events captured during this operation."""
return list(self._domain_events)
# ── Intent-Revealing Methods ────────────────────────────────────────────
def add_item(
self, product_id: str, quantity: int, unit_price: Money
) -> None:
"""Add a line item to this order. Fails fast on invariant violations."""
if self.status != OrderStatus.PENDING:
raise RuntimeError("Cannot modify a non-pending order")
if len(self._items) >= 100:
raise ValueError("Order cannot exceed 100 line items")
item = OrderLineItem(product_id, quantity, unit_price)
self._items.append(item)
self.updated_at = datetime.utcnow()
def remove_item(self, product_id: str) -> None:
"""Remove all line items matching a product ID."""
if self.status != OrderStatus.PENDING:
raise RuntimeError("Cannot modify a non-pending order")
before = len(self._items)
self._items = [i for i in self._items if i.product_id != product_id]
if len(self._items) == before:
raise KeyError(f"Product {product_id} not found in order")
self.updated_at = datetime.utcnow()
self._domain_events.append(OrderItemsChanged(self.id))
def confirm(self, confirmed_by: str) -> None:
"""Transition order to confirmed. Enforces status transition rules."""
if self.status != OrderStatus.PENDING:
raise RuntimeError(
f"Cannot confirm order in {self.status.value} state"
)
if not self._items:
raise ValueError("Cannot confirm an order with no items")
self.status = OrderStatus.CONFIRMED
self.updated_at = datetime.utcnow()
self._domain_events.append(
OrderConfirmed(self.id, self.customer_id, confirmed_by)
)
def cancel(self, reason: str) -> None:
"""Cancel the order with a recorded reason."""
if self.status == OrderStatus.SHIPPED:
raise RuntimeError("Cannot cancel an already shipped order")
if self.status == OrderStatus.CANCELLED:
return # idempotent
was_confirmed = self.status == OrderStatus.CONFIRMED
self.status = OrderStatus.CANCELLED
self.cancelled_at = datetime.utcnow()
self.updated_at = datetime.utcnow()
self._domain_events.append(
OrderCancelled(self.id, reason, was_confirmed)
)
def clear_events(self) -> None:
"""Clear published events after they have been dispatched."""
self._domain_events.clear()
# ── Supporting helpers ────────────────────────────────────────────────────────
from functools import reduce as _reduce
def reduce_money(items): # noqa: D103 — used internally only
return _reduce(lambda a, b: a + b, items, Money(0.0))
# ── BAD Example (violations to avoid) ───────────────────────────────────────
class BadOrderAggregate:
"""❌ BAD examples of aggregate root anti-patterns."""
def __init__(self):
self.items = [] # ❌ Exposes mutable internal collection directly
self.status = "pending" # ❌ Status is a bare string, not an enum
self._events = []
def add_item(self, item): # ❌ No type hints, no validation
self.items.append(item) # ❌ No invariant checks on quantity or price
def get_total(self): # ❌ Returns computed value instead of property
return sum(i.price * i.qty for i in self.items) # ❌ No currency handling
# ── BAD Example: Mutable Value Object ───────────────────────────────────────
class BadMoneyValue:
"""❌ BAD — mutable value object breaks equality semantics."""
def __init__(self, amount: float, currency: str = "USD"):
self.amount = amount # ❌ Mutable attributes
self.currency = currency
def discount(self, percent: float) -> None:
self.amount *= (1 - percent / 100) # ❌ Mutates in place instead of returning new instance
Pattern 2: Domain Events & Synchronous Publishing
Domain events capture facts about state changes. They are published synchronously within the transaction boundary so they commit atomically with aggregate state.
"""Domain event system with synchronous publishing and deduplication."""
from __future__ import annotations
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol
# ── Event Classes ────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class DomainEvent:
"""Base class for all domain events. Immutable and identity-tracked."""
event_id: uuid.UUID = field(default_factory=uuid.uuid4)
occurred_at: datetime = field(default_factory=datetime.utcnow)
@abstractmethod
def aggregate_name(self) -> str: ...
@abstractmethod
def aggregate_id(self) -> str: ...
@dataclass(frozen=True)
class OrderConfirmed(DomainEvent):
event_id: uuid.UUID = field(default_factory=uuid.uuid4)
occurred_at: datetime = field(default_factory=datetime.utcnow)
order_id: str
customer_id: str
confirmed_by: str
def aggregate_name(self) -> str:
return "Order"
def aggregate_id(self) -> str:
return self.order_id
@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
event_id: uuid.UUID = field(default_factory=uuid.uuid4)
occurred_at: datetime = field(default_factory=datetime.utcnow)
order_id: str
reason: str
was_previously_confirmed: bool
def aggregate_name(self) -> str:
return "Order"
def aggregate_id(self) -> str:
return self.order_id
@dataclass(frozen=True)
class OrderItemsChanged(DomainEvent):
event_id: uuid.UUID = field(default_factory=uuid.uuid4)
occurred_at: datetime = field(default_factory=datetime.utcnow)
order_id: uuid.UUID
def aggregate_name(self) -> str:
return "Order"
def aggregate_id(self) -> str:
return str(self.order_id)
# ── Event Publisher Interface & Implementation ───────────────────────────────
class DomainEventPublisher(Protocol):
"""Interface for publishing domain events atomically with a transaction."""
async def publish(self, event: DomainEvent) -> None: ...
async def publish_many(self, events: list[DomainEvent]) -> None: ...
@dataclass
class InMemoryDomainEventPublisher:
"""
Synchronous domain event publisher.
Captures events during aggregate operations and dispatches them within the
same transaction boundary. Uses a change tracker to prevent duplicate
publication of the same event if an operation is retried.
"""
_pending_events: list[DomainEvent] = field(default_factory=list, repr=False)
_published_ids: set[uuid.UUID] = field(default_factory=set, repr=False)
def capture(self, event: DomainEvent) -> None:
"""Queue a domain event for publication."""
self._pending_events.append(event)
def capture_many(self, events: list[DomainEvent]) -> None:
"""Queue multiple domain events for publication."""
self._pending_events.extend(events)
async def publish_pending(self) -> list[DomainEvent]:
"""
Publish all captured events and track them to prevent duplicates.
Returns the list of published events. If called again with already-
published events, they are silently skipped based on event_id dedup.
"""
unpublished = [
ev for ev in self._pending_events if ev.event_id not in self._published_ids
]
for event in unpublished:
await self._do_publish(event)
self._published_ids.add(event.event_id)
# Clear pending events regardless of whether they were new or duplicate
self._pending_events.clear()
return unpublished
async def _do_publish(self, event: DomainEvent) -> None:
"""Publish a single event. Override for real implementations."""
# In production, this would write to an outbox table or message queue.
# The key invariant: publication happens inside the transaction.
pass # noqa: S104 — placeholder for actual persistence logic
def reset(self) -> None:
"""Reset state after a failed transaction roll-back."""
self._pending_events.clear()
# Do NOT clear _published_ids — that event was committed to storage
# ── Unit of Work Coordinator ────────────────────────────────────────────────
class UnitOfWork(ABC):
"""Coordinates transactions: saves aggregates, publishes events atomically."""
@abstractmethod
async def commit(self) -> None: ...
@abstractmethod
async def rollback(self) -> None: ...
class InMemoryUnitOfWork(UnitOfWork):
"""
Implements the unit of work pattern with atomic event publishing.
Saves all registered aggregates and publishes their domain events within
a single transaction boundary. On failure, both state changes and events
are rolled back.
"""
def __init__(self) -> None:
self._aggregates_to_save: list[object] = []
self._publisher = InMemoryDomainEventPublisher()
@property
def publisher(self) -> InMemoryDomainEventPublisher:
return self._publisher
async def commit(self) -> None:
"""Save aggregates and publish events atomically."""
try:
# Step 1: Persist all modified aggregates
for agg in self._aggregates_to_save:
await self._save_aggregate(agg)
# Step 2: Publish all captured domain events (within same txn)
published = await self._publisher.publish_pending()
if published:
# Step 3: Record event persistence to prevent duplicate replay
for ev in published:
self._record_event_published(ev)
except Exception:
await self.rollback()
raise
async def rollback(self) -> None:
"""Rollback all changes including clearing pending events."""
self._aggregates_to_save.clear()
self._publisher.reset()
def register_for_save(self, aggregate: object) -> None:
if aggregate not in self._aggregates_to_save:
self._aggregates_to_save.append(aggregate)
async def _save_aggregate(self, agg: object) -> None:
"""Persist an aggregate root through its repository. Override as needed."""
pass # noqa: S104
def _record_event_published(self, event: DomainEvent) -> None:
"""Record event persistence for deduplication across retries."""
pass # noqa: S104
Pattern 3: Anti-Corruption Layer Adapter
The anti-corruption layer (ACL) translates between external data formats and internal domain models. External model classes must never leak into the domain layer.
"""Anti-corruption layer for translating legacy API responses into domain models."""
from __future__ import annotations
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import Any
# ── Internal Domain Value Objects (these are the ONLY types the domain knows) ─
@dataclass(frozen=True)
class SupplierProduct:
"""Internal domain value object representing a supplier product."""
internal_id: str
sku: str
name: str
price_cents: int
currency: str = "USD"
available_units: int = 0
@dataclass(frozen=True)
class SupplierOrderPayload:
"""Internal domain value object for an incoming order."""
order_reference: str
customer_name: str
items: list[tuple[str, int]] # (sku, quantity)
shipping_address: dict[str, Any]
# ── External API Models (NEVER leak into the domain layer) ───────────────────
@dataclass
class LegacyAPISupplierProduct:
"""External model from the legacy supplier SOAP/REST API. Do NOT use in domain code."""
product_code: str # Maps to internal SKU
supplier_ref: str # Maps to internal_id
description: str # Maps to name
price: float # In dollars — needs conversion to cents
currency_code: str # May vary (USD, EUR, GBP)
stock_quantity: int # Maps to available_units
status: str # Active/Inactive — filtered out
@dataclass
class LegacyAPIOrder:
"""External order format from the legacy system. Do NOT use in domain code."""
po_number: str # Maps to order_reference
customer_name: str | None # Nullable in external, required internally
line_items: list[dict[str, Any]] # Raw dict — must be validated
ship_to: dict[str, str] # May have different keys than internal model
# ── ACL Adapter ──────────────────────────────────────────────────────────────
class SupplierACLAdapter:
"""
Anti-corruption layer adapter for the legacy supplier system.
Translates external API models into internal domain value objects.
All validation and transformation happens here — the domain layer has
zero knowledge of the external data format.
"""
def __init__(self, currency_conversion: dict[str, float] | None = None) -> None:
self._conversion_rates = currency_conversion or {"USD": 1.0}
def map_product(self, external: LegacyAPISupplierProduct) -> SupplierProduct:
"""
Transform a single legacy API product into an internal domain value object.
Raises ValueError if required fields are missing or invalid.
The domain layer will never see LegacyAPISupplierProduct.
"""
if not external.supplier_ref:
raise ValueError("Supplier reference is required")
if not external.product_code:
raise ValueError("Product code (SKU) is required")
if external.status != "Active":
raise ValueError(f"Cannot import inactive product: {external.supplier_ref}")
# Convert price from dollars to cents
rate = self._conversion_rates.get(external.currency_code, 1.0)
price_cents = int(external.price * rate * 100)
return SupplierProduct(
internal_id=external.supplier_ref.strip(),
sku=external.product_code.strip().upper(),
name=external.description.strip(),
price_cents=max(price_cents, 0), # Enforce non-negative at boundary
currency=external.currency_code,
available_units=external.stock_quantity,
)
def map_products(self, external_list: list[LegacyAPISupplierProduct]) -> list[SupplierProduct]:
"""Transform a batch of products. Some may be rejected."""
results: list[SupplierProduct] = []
for ext in external_list:
try:
results.append(self.map_product(ext))
except ValueError:
# Skip inactive or malformed products — do not fail the whole batch
continue
return results
def map_order(self, external: LegacyAPIOrder) -> SupplierOrderPayload:
"""
Transform a legacy API order into an internal domain value object.
Validates all required fields and normalizes the data format.
Raises ValueError on invalid orders before they reach the domain layer.
"""
if not external.po_number:
raise ValueError("Purchase order number is required")
if not external.customer_name:
raise ValueError(f"Customer name required for PO {external.po_number}")
validated_items: list[tuple[str, int]] = []
for raw_item in external.line_items:
sku = raw_item.get("sku", "").strip().upper()
qty = raw_item.get("quantity", 0)
if not sku or qty <= 0:
raise ValueError(f"Invalid line item in PO {external.po_number}: {raw_item}")
validated_items.append((sku, qty))
if not validated_items:
raise ValueError(f"PO {external.po_number} has no valid line items")
return SupplierOrderPayload(
order_reference=external.po_number.strip(),
customer_name=external.customer_name.strip(),
items=validated_items,
shipping_address={
"street": external.ship_to.get("address1", ""),
"city": external.ship_to.get("city", ""),
"state": external.ship_to.get("state", ""),
"postal_code": external.ship_to.get("zip", ""),
"country": external.ship_to.get("country", "US"),
},
)
# ── BAD Example: ACL Anti-Patterns ───────────────────────────────────────────
class BadACLEntryPoint:
"""❌ BAD — leaks external model directly into the domain layer."""
def process_order(self, raw_api_data: dict) -> None:
# ❌ Domain code receives raw API dictionaries
order = ExternalOrder(**raw_api_data) # ❌ Foreign class enters domain
self.domain_service.create(order) # ❌ Domain method takes foreign type
def sync_products(self, products: list[ExternalProduct]) -> None:
# ❌ No translation — external models used directly in queries
for ext_product in products:
session.query(InternalProduct).filter_by(sku=ext_product.product_code) # ❌ Skips ACL
# ── GOOD Example: Clean ACL Boundary ─────────────────────────────────────────
class GoodACLEntryPoint:
"""✅ GOOD — external models are fully translated before reaching the domain."""
def __init__(self, adapter: SupplierACLAdapter) -> None:
self._adapter = adapter
def process_order(self, raw_api_xml: bytes) -> None:
# Step 1: Parse external format (outside domain)
tree = ET.fromstring(raw_api_xml)
external = LegacyAPIOrder(
po_number=tree.findtext("po_number"),
customer_name=tree.findtext("customer_name"),
line_items=[{"sku": i.findtext("sku"), "quantity": int(i.findtext("qty"))}
for i in tree.findall("item")],
ship_to={k: tree.findtext(f"ship_{k}")
for k in ("address1", "city", "state", "zip", "country")},
)
# Step 2: Translate into domain value objects via ACL
payload = self._adapter.map_order(external)
# Step 3: Pass only internal domain types to the domain layer
self.domain_service.create_order(
order_ref=payload.order_reference,
customer_name=payload.customer_name,
items=payload.items,
)
Pattern 4: Repository with Optimistic Concurrency
Repositories manage aggregate root persistence with optimistic concurrency control. Read queries use separate projection readers.
"""Repository pattern for aggregate roots with optimistic concurrency control."""
from __future__ import annotations
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Generic, Iterator, Protocol, TypeVar
# ── Optimistic Concurrency Base ──────────────────────────────────────────────
@dataclass
class VersionedAggregate:
"""
Base class providing optimistic concurrency control.
Each aggregate carries a version number that increments on every save.
Concurrent modifications are detected when the stored version differs
from the expected version at commit time.
"""
id: uuid.UUID
version: int = 0
updated_at: datetime = field(default_factory=datetime.utcnow)
def bump_version(self) -> None:
self.version += 1
self.updated_at = datetime.utcnow()
# ── Specification Pattern ───────────────────────────────────────────────────
class Specification(Protocol):
"""Protocol for composable business rule specifications."""
def is_satisfied_by(self, candidate: object) -> bool: ...
def and_spec(self, other: Specification) -> Specification: ...
def or_spec(self, other: Specification) -> Specification: ...
def not_spec(self) -> Specification: ...
class CompositeSpecification(Specification):
"""Composable specification using AND/OR/NOT composition."""
def __init__(self, left: Specification | None = None, right: Specification | None = None,
operator: str = "and") -> None:
self._left = left
self._right = right
self._operator = operator
@classmethod
def always(cls) -> Specification:
"""Specification that always returns True."""
return _AlwaysSpec()
@classmethod
def never(cls) -> Specification:
"""Specification that always returns False."""
return _NeverSpec()
def is_satisfied_by(self, candidate: object) -> bool:
if self._left is None and self._right is None:
return True # Leaf specs override
left_ok = self._left.is_satisfied_by(candidate) if self._left else True
if self._operator == "and":
right_ok = self._right.is_satisfied_by(candidate) if self._right else True
return left_ok and right_ok
elif self._operator == "or":
right_ok = self._right.is_satisfied_by(candidate) if self._right else False
return left_ok or right_ok
raise RuntimeError(f"Unknown operator: {self._operator}")
def and_spec(self, other: Specification) -> Specification:
return CompositeSpecification(self, other, "and")
def or_spec(self, other: Specification) -> Specification:
return CompositeSpecification(self, other, "or")
def not_spec(self) -> Specification:
return _NegatedSpec(self)
class _AlwaysSpec(Specification):
def is_satisfied_by(self, candidate: object) -> bool: return True
def and_spec(self, other: Specification) -> Specification: return other
def or_spec(self, other: Specification) -> Specification: return self
def not_spec(self) -> Specification: return _NeverSpec()
class _NeverSpec(Specification):
def is_satisfied_by(self, candidate: object) -> bool: return False
def and_spec(self, other: Specification) -> Specification: return self
def or_spec(self, other: Specification) -> Specification: return other
def not_spec(self) -> Specification: return _AlwaysSpec()
class _NegatedSpec(Specification):
def __init__(self, spec: Specification) -> None:
self._spec = spec
def is_satisfied_by(self, candidate: object) -> bool:
return not self._spec.is_satisfied_by(candidate)
def and_spec(self, other: Specification) -> Specification:
return CompositeSpecification(self._spec.not_spec(), other, "and")
def or_spec(self, other: Specification) -> Specification:
return CompositeSpecification(self._spec.not_spec(), other, "or")
def not_spec(self) -> Specification:
return self._spec # Double negation
class OrderStatusSpec(Specification):
"""Filters by order status."""
def __init__(self, status: str) -> None:
self._status = status
def is_satisfied_by(self, candidate: object) -> bool:
return hasattr(candidate, "status") and candidate.status == self._status
class MinTotalSpec(Specification):
"""Filters by minimum total amount."""
def __init__(self, minimum_cents: int) -> None:
self._minimum_cents = minimum_cents
def is_satisfied_by(self, candidate: object) -> bool:
if not hasattr(candidate, "total"):
return False
# Assumes `total` has a `cents` property or can be converted
total_val = getattr(getattr(candidate, "total", None), "amount", 0) * 100
return total_val >= self._minimum_cents
class ActiveCustomerSpec(Specification):
"""Filters by whether the customer is active."""
def __init__(self, get_customer_status) -> None: # Injection point for external check
self._get_customer_status = get_customer_status
def is_satisfied_by(self, candidate: object) -> bool:
if not hasattr(candidate, "customer_id"):
return False
status = self._get_customer_status(candidate.customer_id)
return status == "active"
# ── Repository Protocol & Implementation ─────────────────────────────────────
T = TypeVar("T", bound=VersionedAggregate)
class Repository(Protocol, Generic[T]):
"""Repository interface for an aggregate root type."""
@abstractmethod
async def load(self, id: uuid.UUID) -> T | None: ...
@abstractmethod
async def save(self, aggregate: T) -> None: ...
@abstractmethod
async def find_by(self, spec: Specification) -> list[T]: ...
class InMemoryOrderRepository:
"""
In-memory repository demonstrating optimistic concurrency control.
Production implementations would replace the in-memory store with a real
database. The key pattern: every save checks the version number, and a
mismatch raises ConcurrencyConflictError.
"""
def __init__(self) -> None:
self._store: dict[uuid.UUID, VersionedAggregate] = {}
self._on_save_hooks: list[callable] = []
async def load(self, id: uuid.UUID) -> Order | None:
"""Load an aggregate root by ID. Returns None if not found."""
return self._store.get(id) # type: ignore[return-value]
async def save(self, aggregate: VersionedAggregate) -> None:
"""
Save an aggregate with optimistic concurrency control.
Raises ConcurrencyConflictError if the aggregate's version does not
match the stored version, indicating a concurrent modification.
"""
existing = self._store.get(aggregate.id)
if existing is not None and existing.version != aggregate.version:
raise ConcurrencyConflictError(
f"Aggregate {aggregate.id}: expected version {aggregate.version}, "
f"but stored version is {existing.version}. Another process modified it."
)
# Apply any pre-save hooks (e.g., set updated_at, audit fields)
for hook in self._on_save_hooks:
hook(aggregate)
aggregate.bump_version()
self._store[aggregate.id] = aggregate
async def find_by(self, spec: Specification) -> list[VersionedAggregate]:
"""Find all aggregates matching a specification."""
return [agg for agg in self._store.values() if spec.is_satisfied_by(agg)] # type: ignore[arg-type]
def register_save_hook(self, hook: callable) -> None:
self._on_save_hooks.append(hook)
class ConcurrencyConflictError(Exception):
"""Raised when optimistic concurrency check fails."""
def __init__(self, message: str) -> None:
super().__init__(message)
self.is_retryable = True # Signal that the caller should retry
# ── Read-Side Projection (Separate from Write Model) ────────────────────────
@dataclass
class OrderProjection:
"""Read-side projection — a flattened, query-optimized view of order data."""
order_id: str
customer_id: str
status: str
item_count: int
total_cents: int
created_at: datetime
class OrderProjectionReader:
"""
Separate reader for queries that don't modify state.
This keeps read concerns completely separate from the write model.
Projections can be denormalized, indexed differently, and queried
independently without touching aggregate roots.
"""
def __init__(self) -> None:
self._projections: dict[str, OrderProjection] = {}
def apply_event(self, event: DomainEvent) -> None:
"""Rebuild projections from domain events (eventual consistency)."""
if isinstance(event, OrderConfirmed):
# In production, this would load the aggregate and project it
self._projections[event.order_id] = OrderProjection(
order_id=event.order_id,
customer_id=event.customer_id,
status="confirmed",
item_count=0, # Would come from actual items
total_cents=0, # Would be computed
created_at=datetime.utcnow(),
)
def get_by_customer(self, customer_id: str) -> list[OrderProjection]:
return [p for p in self._projections.values() if p.customer_id == customer_id]
def find_pending(self) -> list[OrderProjection]:
return [p for p in self._projections.values() if p.status == "pending"]
Pattern 5: Specification Composition & Query Objects
Specifications are composable business rules that work both for domain validation and repository filtering. Query objects encapsulate complex read queries.
"""Composable specifications and query objects for read-side operations."""
from __future__ import annotations
from dataclasses import dataclass, field
# ── Composable Specifications (continued from Pattern 4) ────────────────────
def create_order_query_spec(
min_total_cents: int | None = None,
exclude_statuses: list[str] | None = None,
require_active_customer: bool = False,
) -> Specification:
"""
Factory for building complex order filtering specifications.
Demonstrates specification composition — combine simple specs with
AND/OR/NOT to build arbitrarily complex business rules.
Example: "Show me orders over $50 from active customers that are pending or confirmed"
spec = create_order_query_spec(min_total_cents=5000, exclude_statuses=["cancelled", "shipped"])
"""
base: Specification = Specification.always()
if min_total_cents is not None:
base = base.and_spec(MinTotalSpec(min_total_cents))
if exclude_statuses:
# NOT any of the excluded statuses — compose by negating each
exclusion_specs = [OrderStatusSpec(s) for s in exclude_statuses]
combined_exclusion = Specification.always()
for spec_obj in exclusion_specs:
combined_exclusion = combined_exclusion.or_spec(spec_obj)
base = base.and_spec(combined_exclusion.not_spec())
if require_active_customer:
base = base.and_spec(ActiveCustomerSpec(get_customer_status=lambda cid: "active"))
return base
# ── Query Object Pattern ────────────────────────────────────────────────────
@dataclass(frozen=True)
class OrderQuery:
"""
Immutable query object for complex read operations.
Encapsulates all parameters needed to build a query without touching
the write model. Can be converted to SQL WHERE clauses, cached, and
reused across repositories.
"""
customer_id: str | None = None
status_filter: list[str] | None = None
date_range_start: datetime | None = None
date_range_end: datetime | None = None
min_total_cents: int | None = None
max_total_cents: int | None = None
product_sku: str | None = None
page: int = 1
page_size: int = 50
@property
def limit(self) -> int:
return min(self.page_size, 100)
@property
def offset(self) -> int:
return (self.page - 1) * self.limit
def to_where_clause(self) -> tuple[str, list[object]]:
"""
Convert this query object into a SQL WHERE clause and parameters.
Returns a tuple of (clause_string, params_list) ready for parameterized queries.
This is how specifications bridge between domain logic and persistence.
"""
conditions: list[str] = []
params: list[object] = []
if self.customer_id:
conditions.append("customer_id = ?")
params.append(self.customer_id)
if self.status_filter:
placeholders = ", ".join(["?"] * len(self.status_filter))
conditions.append(f"status IN ({placeholders})")
param
…(truncated)