Ddd Modeling

Domain-Driven Design (DDD) Modeler

Purpose

Model complex business domains using DDD tactical and strategic patterns to create software that reflects real business logic.

Strategic DDD

1️⃣ Domain Discovery (Event Storming)

Run a collaborative session:

1. Domain Events (orange): What happened? (OrderPlaced, PaymentFailed, ItemShipped)
2. Commands (blue): What triggers events? (PlaceOrder, ProcessPayment)
3. Actors (yellow): Who sends commands? (Customer, System, Admin)
4. Aggregates (yellow): What handles commands? (Order, Payment)
5. Policies (purple): When X happens, do Y
6. External Systems (pink): Third-party integrations

2️⃣ Bounded Contexts

Identify distinct business areas with their own ubiquitous language:

Order Management Context: Order, OrderItem, Fulfillment, Shipment
Payment Context: Payment, Transaction, Refund, Invoice
Inventory Context: Product, Stock, Warehouse, Reservation
Customer Context: Customer, Account, Address, Preferences

3️⃣ Aggregate Design

# Aggregate root enforces business invariants
class Order(AggregateRoot):
    _MAX_ITEMS = 50
    
    def __init__(self, order_id: OrderId, customer_id: CustomerId):
        self._id = order_id
        self._customer_id = customer_id
        self._items: list[OrderItem] = []
        self._status = OrderStatus.DRAFT
        self._events: list[DomainEvent] = []
    
    def add_item(self, product_id: ProductId, quantity: int, price: Money) -> None:
        # Enforce business rules (invariants)
        if self._status != OrderStatus.DRAFT:
            raise OrderNotModifiableError("Cannot add items to a placed order")
        if len(self._items) >= self._MAX_ITEMS:
            raise OrderLimitExceededError(f"Maximum {self._MAX_ITEMS} items allowed")
        
        # Update state
        existing = self._find_item(product_id)
        if existing:
            existing.increase_quantity(quantity)
        else:
            self._items.append(OrderItem(product_id, quantity, price))
        
        # Raise domain event
        self._raise(ItemAddedToOrder(self._id, product_id, quantity))
    
    def place(self) -> None:
        if not self._items:
            raise EmptyOrderError("Cannot place an empty order")
        
        self._status = OrderStatus.PLACED
        self._raise(OrderPlaced(
            order_id=self._id,
            customer_id=self._customer_id,
            total=self._calculate_total(),
            placed_at=datetime.utcnow(),
        ))
    
    def _raise(self, event: DomainEvent) -> None:
        self._events.append(event)

4️⃣ Domain Events

@dataclass(frozen=True)
class OrderPlaced:
    order_id: OrderId
    customer_id: CustomerId
    total: Money
    placed_at: datetime
    
    # Used for routing to correct handler
    @property
    def event_type(self) -> str:
        return "order.placed"

# Event handlers in other contexts
class PaymentContext:
    def handle_order_placed(self, event: OrderPlaced) -> None:
        # Create a pending payment in payment context
        payment = Payment.create_for_order(event.order_id, event.total)
        self.payments.save(payment)

5️⃣ Repository Pattern

class OrderRepository(Protocol):
    def find_by_id(self, order_id: OrderId) -> Order | None: ...
    def save(self, order: Order) -> None: ...  # Also dispatches domain events
    def find_by_customer(self, customer_id: CustomerId) -> list[Order]: ...

class PostgresOrderRepository:
    def save(self, order: Order) -> None:
        # Persist the aggregate
        with self.db.transaction():
            self.db.execute(...)
            # Dispatch domain events after successful save
            for event in order.collect_events():
                self.event_bus.publish(event)

Outputs

1. Bounded context map
2. Ubiquitous language glossary
3. Aggregate design with invariants
4. Domain event catalog
5. Context integration patterns (ACL, Published Language)