Loading...
Loading...
DDD aggregate design patterns for consistency boundaries and invariants. Use when designing aggregate roots, enforcing business invariants, handling cross-aggregate references, or optimizing aggregate size.
npx skill4agent add yonatangross/orchestkit aggregate-patterns┌─────────────────────────────────────────────────────────┐
│ ORDER AGGREGATE │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Order (Aggregate Root) │ │
│ │ • id: UUID (UUIDv7) │ │
│ │ • customer_id: UUID (reference by ID!) │ │
│ │ • status: OrderStatus │ │
│ └─────────────────────────────────────────────────┘ │
│ │ │ │
│ ┌────────────────┐ ┌────────────────┐ │
│ │ OrderItem │ │ OrderItem │ │
│ │ (child) │ │ (child) │ │
│ └────────────────┘ └────────────────┘ │
│ │
│ INVARIANTS enforced by root: │
│ • Total = sum of items │
│ • Max 100 items per order │
│ • Cannot modify after shipped │
└─────────────────────────────────────────────────────────┘from dataclasses import dataclass, field
from uuid import UUID
from uuid_utils import uuid7
@dataclass
class OrderAggregate:
"""Aggregate root with invariant enforcement."""
id: UUID = field(default_factory=uuid7)
customer_id: UUID # Reference by ID, not Customer object!
_items: list["OrderItem"] = field(default_factory=list)
status: str = "draft"
MAX_ITEMS = 100
def add_item(self, product_id: UUID, quantity: int, price: Money) -> None:
"""Add item with invariant checks."""
self._ensure_modifiable()
if len(self._items) >= self.MAX_ITEMS:
raise DomainError("Max items exceeded")
self._items.append(OrderItem(product_id, quantity, price))
def _ensure_modifiable(self) -> None:
if self.status != "draft":
raise DomainError(f"Cannot modify {self.status} order")| Decision | Recommendation |
|---|---|
| Aggregate size | Small (< 20 children), split if larger |
| Cross-aggregate refs | Always by ID, never by object |
| Consistency | Immediate within, eventual across |
| Events | Collect in root, publish after persist |
# NEVER reference aggregates by object
customer: Customer # WRONG → customer_id: UUID
# NEVER modify multiple aggregates in one transaction
order.submit()
inventory.reserve(items) # WRONG - use domain events
# NEVER expose mutable collections
def items(self) -> list:
return self._items # WRONG → return tuple(self._items)
# NEVER have unbounded collections
orders: list[Order] # WRONG - grows unboundeddomain-driven-designdistributed-locksidempotency-patterns