DDD in Action: From Sticky Notes to Production Code¶

Theory is easy. Practice is where DDD gets real — and where most teams get stuck.

You've read about Bounded Contexts and Aggregates. You understand Ubiquitous Language. But when you sit down with your team in front of a real codebase and real business requirements, a critical question remains: how do you actually do this?

This guide takes you through the complete DDD journey on a concrete real-world system — an online bookstore — from the first collaborative workshop all the way to working production code. Every design decision is explained. Every mistake is shown before the fix. By the end, you'll have a repeatable playbook you can apply to your own system.

This is Part 2 of a DDD series

This guide focuses on the practice of applying DDD. For the foundational theory — Bounded Contexts, Aggregates, Value Objects, Domain Events — read the DDD Complete Guide first.

Our Running Example: An Online Bookstore¶

Throughout this guide we'll build the core of an online bookstore called Bookly. It sells physical and digital books, supports memberships with discounts, and ships orders worldwide.

At first glance it seems simple. But as we'll discover through Event Storming, the domain is richer and more complex than it appears — exactly the kind of system where DDD pays off.

The Journey Overview¶

DDD in action follows six phases. We'll cover each one in depth:

  Phase 1          Phase 2          Phase 3
  DISCOVER    →    DEFINE      →    DESIGN
  (Event            (Bounded         (Aggregates
   Storming)         Contexts)        & Models)
       │
       │
  Phase 6          Phase 5          Phase 4
  VALIDATE    ←    INTEGRATE   ←    IMPLEMENT
  (Tests &          (Events &        (Code the
   Refinement)       APIs)            Domain)

Phase 1: Domain Discovery with Event Storming¶

Event Storming is the fastest way to build shared understanding of a complex domain. It was invented by Alberto Brandolini and takes 4–8 hours for a domain the size of Bookly.

Who Needs to Be in the Room¶

This is not a developer-only session. You need everyone who understands a slice of the business:

Role	What They Contribute
Domain Experts	What actually happens in the business
Product Managers	What outcomes the business needs
Developers	What's technically feasible
UX Designers	What users experience
Support / Ops	What goes wrong and how it's fixed

The most valuable moments in Event Storming happen when a developer says "wait, that's not how I thought this worked" and a domain expert says "yes, that's exactly the problem we've had for years."

Setting Up the Room¶

You need:

A wall at least 5 metres wide (use paper rolls taped together)
Sticky notes in six colours — buy 300+ of each
Thick markers (Sharpies) — thin pens can't be read from a distance
Roughly 2 hours per session (4 sessions for a full domain)

Colour	Represents	Format
🟠 Orange	Domain Events	Past tense verb — "Order Placed"
🔵 Blue	Commands	Imperative verb — "Place Order"
🟡 Yellow	Actors / Users	Who triggers the command
🟣 Purple	Policies	"When X happens, do Y"
🩷 Pink	External Systems	Stripe, SendGrid, FedEx API
🟢 Green	Read Models	What information a user sees

Step 1: Chaotic Exploration (45 minutes)¶

Everyone writes Domain Events on orange stickies and places them on the wall — in any order, no rules. The facilitator's only job is to keep the energy up.

Events are things that happened in the business — past tense, business language, meaningful to a domain expert:

🟠 Book Searched        🟠 Book Added to Cart     🟠 Checkout Started
🟠 Payment Processed    🟠 Order Confirmed         🟠 Inventory Reserved
🟠 Shipment Dispatched  🟠 Book Downloaded         🟠 Membership Activated
🟠 Review Submitted     🟠 Refund Requested        🟠 Order Cancelled
🟠 Delivery Confirmed   🟠 Discount Applied        🟠 Wishlist Updated
🟠 Coupon Redeemed      🟠 Address Validated       🟠 Fraud Detected

The 'stupid question' rule

During chaotic exploration, no sticky note is wrong. Encourage everyone to write duplicates — duplicates reveal disagreements about terminology, which is exactly what you need to surface.

Step 2: Order the Timeline (30 minutes)¶

Arrange the events left-to-right in rough chronological order. Don't force perfect order — focus on revealing the narrative:

  TIMELINE (left → right = earlier → later):

  ──────────────────────────────────────────────────────────────────→ time

  [Book       [Book Added   [Checkout    [Payment     [Order       [Book
   Searched]   to Cart]      Started]     Processed]   Confirmed]   Downloaded]

                                              ↕
                                         [Payment
                                          Failed]

  [Order       [Inventory   [Shipment    [Delivery    [Review
   Confirmed]   Reserved]    Dispatched]  Confirmed]   Submitted]

Disagreements about ordering are gold — they reveal missing steps, unclear ownership, and hidden complexity.

Step 3: Commands and Actors (30 minutes)¶

For each event, add the blue Command that caused it and the yellow Actor who triggered it:

  🟡 Customer     🔵 Search Books    →    🟠 Book Searched
  🟡 Customer     🔵 Add to Cart     →    🟠 Book Added to Cart
  🟡 Customer     🔵 Initiate Checkout → 🟠 Checkout Started
  🟡 Payment GW   🔵 Process Payment →   🟠 Payment Processed
  🟣 Policy       [When Payment OK]  →    🟠 Order Confirmed
  🟣 Policy       [When Order OK]    →    🟠 Inventory Reserved
  🟡 Warehouse    🔵 Dispatch Order  →    🟠 Shipment Dispatched
  🟡 System       🔵 Deliver File    →    🟠 Book Downloaded

Step 4: Aggregates and Policies (30 minutes)¶

Group commands and events under yellow Aggregate stickies (large ones). Add purple Policy stickies that describe automation rules:

  🟡 CART Aggregate:
     [Add to Cart] → [Book Added to Cart]
     [Remove from Cart] → [Book Removed from Cart]
     [Apply Coupon] → [Coupon Applied]

  🟡 ORDER Aggregate:
     [Place Order] → [Order Placed]
     [Cancel Order] → [Order Cancelled]
     [Confirm Order] ← 🟣 "When payment succeeds"

  🟡 PAYMENT Aggregate:
     [Process Payment] → [Payment Processed]
                      → [Payment Failed]
     [Refund Payment] → [Refund Issued]

  🟣 POLICY: When Order Confirmed → Reserve Inventory
  🟣 POLICY: When Payment Failed → Restore Cart Items
  🟣 POLICY: When Inventory Reserved → Schedule Shipment

Step 5: External Systems (20 minutes)¶

Add pink stickies for systems your domain interacts with:

  🩷 Stripe           (payment processing)
  🩷 SendGrid         (email notifications)
  🩷 FedEx API        (shipping rates + tracking)
  🩷 Google Books API (book metadata enrichment)
  🩷 Fraud Detection  (order risk scoring)

Reading the Event Storm Output¶

When you step back, you should see natural groupings — areas of the board that cluster tightly. These are your candidate Bounded Contexts:

  ┌─────────────────┐  ┌──────────────┐  ┌──────────────────┐
  │ CATALOG cluster │  │ ORDER cluster│  │ PAYMENT cluster  │
  │                 │  │              │  │                  │
  │ Book Searched   │  │ Cart Updated │  │ Payment Process  │
  │ Catalog Updated │  │ Order Placed │  │ Fraud Checked    │
  │ Review Added    │  │ Order Cancel │  │ Refund Issued    │
  └─────────────────┘  └──────────────┘  └──────────────────┘

  ┌──────────────────┐  ┌───────────────┐  ┌──────────────┐
  │ INVENTORY cluster│  │SHIPPING cluster│  │ MEMBER cluster│
  │                  │  │                │  │              │
  │ Stock Reserved   │  │ Shipment Sched │  │ Member Joined│
  │ Stock Released   │  │ Label Printed  │  │ Points Earned│
  │ Stock Adjusted   │  │ Delivered      │  │ Tier Upgraded│
  └──────────────────┘  └───────────────┘  └──────────────┘

Phase 2: Defining Bounded Contexts¶

With the Event Storm output in hand, we now draw formal boundaries.

The Six Bounded Contexts of Bookly¶

  ┌───────────────────────────────────────────────────────────────────┐
  │                        BOOKLY PLATFORM                            │
  │                                                                   │
  │  ┌──────────┐   ┌──────────┐   ┌──────────┐   ┌──────────────┐  │
  │  │ CATALOG  │   │  ORDERS  │   │ PAYMENTS │   │  INVENTORY   │  │
  │  │          │   │          │   │          │   │              │  │
  │  │Books     │   │Cart      │   │Transact  │   │Stock         │  │
  │  │Authors   │   │Orders    │   │Refunds   │   │Reservations  │  │
  │  │Reviews   │   │Returns   │   │Invoices  │   │Adjustments   │  │
  │  └──────────┘   └──────────┘   └──────────┘   └──────────────┘  │
  │                                                                   │
  │  ┌──────────────┐   ┌─────────────────────────────────────────┐  │
  │  │   SHIPPING   │   │           MEMBERSHIP                    │  │
  │  │              │   │                                         │  │
  │  │Shipments     │   │Members  Tiers  Points  Discounts        │  │
  │  │Tracking      │   │                                         │  │
  │  │Carriers      │   └─────────────────────────────────────────┘  │
  │  └──────────────┘                                                 │
  └───────────────────────────────────────────────────────────────────┘

The Same Word, Different Worlds¶

Here's how the term "Book" means something different in each context:

Context	"Book" means	Key attributes
Catalog	Something a customer can discover	Title, author, description, cover image, reviews, ISBN
Orders	A line item in a purchase	Product ID, title (snapshot), price at time of order
Inventory	A unit of stock	SKU, quantity on hand, reorder threshold, warehouse location
Shipping	A physical package	Weight, dimensions, fragility, shipping class

This is intentional, not a design flaw

Seeing the same concept modeled differently in each context is correct DDD. Forcing a single shared Book model across all contexts creates a tightly coupled, fragile system. Separate models, even with some duplication, are the right call.

Building the Context Map¶

Now document how the bounded contexts relate to each other:

  BOOKLY CONTEXT MAP

  ┌──────────┐                              ┌──────────┐
  │ CATALOG  │ ──── Open Host Service ────→ │  ORDERS  │
  │  (U)     │      Published Language      │  (D/ACL) │
  └──────────┘                              └────┬─────┘
                                                 │ Customer-Supplier
                                                 │ (Orders → Payments)
                                                 ▼
                                          ┌──────────┐
                                          │ PAYMENTS │
                                          │          │
                                          │ 🩷 Stripe │
                                          │  (Conform│
                                          │   ist)   │
                                          └──────────┘

  ORDERS ──── Domain Events ────→ INVENTORY  (eventual consistency)
  ORDERS ──── Domain Events ────→ SHIPPING   (eventual consistency)
  ORDERS ──── Domain Events ────→ MEMBERSHIP (eventual consistency)

Why these patterns?

Catalog → Orders (Open Host + ACL): Catalog is a stable reference service. Orders consumes it but translates the catalog Book into its own OrderLineItem through an ACL so catalog changes don't break orders.
Orders → Payments (Customer-Supplier): Orders tells Payments what to charge. Payments has no opinion about the business logic — it just processes money.
Payments → Stripe (Conformist): We conform to Stripe's API. They don't change for us.
Orders → Inventory/Shipping/Membership (Domain Events): Loose coupling via async events — orders doesn't need to know these services exist.

Phase 3: Designing Aggregates¶

We'll focus on the Orders bounded context to show aggregate design in full detail.

The Aggregate Discovery Process¶

Three questions guide aggregate design:

What must be consistent together? (consistency boundary)
What is the entry point for all changes? (aggregate root)
What can be eventually consistent? (use events, not direct references)

Candidate Aggregates in the Orders Context¶

From our Event Storm, we have these events in the Orders context:

Cart:     Book Added to Cart, Book Removed from Cart, Coupon Applied,
          Checkout Started, Cart Abandoned
Order:    Order Placed, Order Confirmed, Order Cancelled,
          Return Requested, Return Approved

This suggests two aggregates: Cart and Order.

Why Not One Big "Purchase" Aggregate?¶

A beginner might model Cart and Order as one Purchase aggregate. Let's see why that's wrong:

# BAD: one giant aggregate — everything locked together
@dataclass
class Purchase:
    id: UUID
    customer_id: UUID
    cart_items: list[CartItem]       # changes frequently
    order_status: OrderStatus        # changes rarely
    payment_info: PaymentInfo        # PCI-sensitive data
    shipping_address: Address
    return_requests: list[Return]    # different lifecycle
    invoices: list[Invoice]          # finance team owns this

Problems: - A customer adding a book to cart locks the entire Purchase record, blocking a support agent from processing a return - Loading an order for status display forces loading all cart history - Different teams need to own different parts — impossible if they share one aggregate

The Correct Model: Two Small Aggregates¶

# GOOD: small, focused aggregates
@dataclass
class Cart:                          # Aggregate Root
    id: UUID
    customer_id: UUID
    _items: list[CartItem]
    _coupon: Optional[Coupon]
    # Cart cares about: items, totals, coupon validity

@dataclass
class Order:                         # Aggregate Root
    id: UUID
    customer_id: UUID
    cart_id: UUID                    # reference by ID only — not the object
    _line_items: list[OrderLineItem]
    _status: OrderStatus
    _shipping_address: Address
    # Order cares about: status transitions, line items (snapshot), shipping

Cart and Order are linked by cart_id — an ID reference, not an object reference. When a Cart is converted to an Order, the Order takes a snapshot of the cart's contents. From that point on, the Cart and Order evolve independently.

Phase 4: Implementing the Domain in Code¶

Let's build the complete Orders bounded context. We'll use Python, but every pattern applies to any language.

Folder Structure¶

orders/
├── domain/
│   ├── __init__.py
│   ├── cart.py           # Cart aggregate
│   ├── order.py          # Order aggregate
│   ├── value_objects.py  # Money, Address, Coupon, etc.
│   ├── events.py         # Domain events
│   └── exceptions.py     # Domain exceptions
├── application/
│   ├── __init__.py
│   ├── cart_service.py   # Use cases for cart
│   ├── order_service.py  # Use cases for orders
│   └── dto.py            # Data Transfer Objects
├── infrastructure/
│   ├── __init__.py
│   ├── postgres_cart_repo.py
│   ├── postgres_order_repo.py
│   └── event_publisher.py
└── api/
    ├── __init__.py
    ├── cart_router.py
    └── order_router.py

Value Objects¶

Value objects carry meaning and enforce constraints without an identity:

# domain/value_objects.py
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from uuid import UUID


@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError(f"Money amount cannot be negative: {self.amount}")
        if len(self.currency) != 3:
            raise ValueError(f"Currency must be ISO 3-letter code: {self.currency}")
        # Normalize to 2 decimal places
        object.__setattr__(self, 'amount', self.amount.quantize(
            Decimal('0.01'), rounding=ROUND_HALF_UP
        ))

    def add(self, other: 'Money') -> 'Money':
        self._assert_same_currency(other)
        return Money(self.amount + other.amount, self.currency)

    def subtract(self, other: 'Money') -> 'Money':
        self._assert_same_currency(other)
        result = self.amount - other.amount
        if result < 0:
            raise ValueError("Result would be negative")
        return Money(result, self.currency)

    def multiply(self, factor: Decimal) -> 'Money':
        return Money(self.amount * factor, self.currency)

    def _assert_same_currency(self, other: 'Money') -> None:
        if self.currency != other.currency:
            raise ValueError(
                f"Currency mismatch: {self.currency} vs {other.currency}"
            )

    def __str__(self) -> str:
        return f"{self.amount} {self.currency}"

    @classmethod
    def zero(cls, currency: str = "USD") -> 'Money':
        return cls(Decimal("0.00"), currency)


@dataclass(frozen=True)
class BookId:
    value: UUID

    def __str__(self):
        return str(self.value)


@dataclass(frozen=True)
class CustomerId:
    value: UUID

    def __str__(self):
        return str(self.value)


@dataclass(frozen=True)
class Address:
    street_line_1: str
    street_line_2: str
    city: str
    state_or_province: str
    postal_code: str
    country_code: str  # ISO 3166-1 alpha-2

    def __post_init__(self):
        if not self.street_line_1.strip():
            raise ValueError("Street address is required")
        if len(self.country_code) != 2:
            raise ValueError("Country code must be ISO 2-letter code")

    @property
    def is_international(self) -> bool:
        return self.country_code != "US"


@dataclass(frozen=True)
class Coupon:
    code: str
    discount_percent: int  # 0-100

    def __post_init__(self):
        if not (0 < self.discount_percent <= 100):
            raise ValueError(f"Discount must be 1-100%: {self.discount_percent}")
        if not self.code.strip():
            raise ValueError("Coupon code cannot be empty")

    def apply_to(self, amount: Money) -> Money:
        discount = amount.multiply(Decimal(self.discount_percent) / 100)
        return amount.subtract(discount)

Domain Events¶

Events are immutable records of things that happened:

# domain/events.py
from dataclasses import dataclass, field
from datetime import datetime, UTC
from uuid import UUID, uuid4
from .value_objects import Money


@dataclass(frozen=True)
class DomainEvent:
    event_id: UUID = field(default_factory=uuid4)
    occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC))


@dataclass(frozen=True)
class BookAddedToCart(DomainEvent):
    cart_id: UUID
    customer_id: UUID
    book_id: UUID
    book_title: str
    unit_price: Money
    quantity: int


@dataclass(frozen=True)
class BookRemovedFromCart(DomainEvent):
    cart_id: UUID
    book_id: UUID


@dataclass(frozen=True)
class CouponApplied(DomainEvent):
    cart_id: UUID
    coupon_code: str
    discount_percent: int


@dataclass(frozen=True)
class CartCheckedOut(DomainEvent):
    cart_id: UUID
    customer_id: UUID
    total_amount: Money


@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: UUID
    customer_id: UUID
    cart_id: UUID
    total_amount: Money
    item_count: int


@dataclass(frozen=True)
class OrderConfirmed(DomainEvent):
    order_id: UUID
    customer_id: UUID
    total_amount: Money


@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
    order_id: UUID
    customer_id: UUID
    reason: str
    refund_amount: Money


@dataclass(frozen=True)
class ReturnRequested(DomainEvent):
    order_id: UUID
    customer_id: UUID
    reason: str

Domain Exceptions¶

Use specific exceptions that speak the Ubiquitous Language:

# domain/exceptions.py

class DomainError(Exception):
    """Base class for all domain errors."""


class EmptyCartError(DomainError):
    """Cannot check out an empty cart."""


class BookAlreadyInCartError(DomainError):
    def __init__(self, book_id: str):
        super().__init__(f"Book {book_id} is already in the cart")


class CartItemLimitExceededError(DomainError):
    def __init__(self, limit: int):
        super().__init__(f"Cannot add more than {limit} items to a cart")


class CouponAlreadyAppliedError(DomainError):
    """Only one coupon can be applied per cart."""


class InvalidOrderStatusTransitionError(DomainError):
    def __init__(self, current: str, attempted: str):
        super().__init__(
            f"Cannot transition order from {current} to {attempted}"
        )


class OrderAlreadyConfirmedError(DomainError):
    """Order has already been confirmed and cannot be modified."""


class ReturnWindowExpiredError(DomainError):
    """The 30-day return window has expired for this order."""

The Cart Aggregate¶

# domain/cart.py
from dataclasses import dataclass, field
from decimal import Decimal
from uuid import UUID, uuid4
from typing import Optional

from .value_objects import Money, BookId, CustomerId, Coupon
from .events import (
    BookAddedToCart, BookRemovedFromCart,
    CouponApplied, CartCheckedOut, DomainEvent
)
from .exceptions import (
    BookAlreadyInCartError, CartItemLimitExceededError,
    CouponAlreadyAppliedError, EmptyCartError
)

MAX_CART_ITEMS = 50


@dataclass(frozen=True)
class CartItem:
    """Value Object — a snapshot of a book's details at add-to-cart time."""
    book_id: BookId
    book_title: str
    unit_price: Money
    quantity: int

    @property
    def subtotal(self) -> Money:
        return self.unit_price.multiply(Decimal(self.quantity))


@dataclass
class Cart:
    """
    Aggregate Root for the shopping cart.

    Invariants enforced:
    - Max 50 items
    - A book appears at most once (quantity updated, not duplicated)
    - Only one coupon can be active at a time
    - Cart total is always consistent with items
    """
    id: UUID
    customer_id: CustomerId
    _items: list[CartItem] = field(default_factory=list)
    _coupon: Optional[Coupon] = field(default=None)
    _events: list[DomainEvent] = field(default_factory=list)

    # ── Factory ────────────────────────────────────────────────────
    @classmethod
    def create(cls, customer_id: CustomerId) -> 'Cart':
        return cls(id=uuid4(), customer_id=customer_id)

    # ── Commands ───────────────────────────────────────────────────
    def add_book(
        self,
        book_id: BookId,
        book_title: str,
        unit_price: Money,
        quantity: int = 1
    ) -> None:
        # Enforce: max items limit
        if len(self._items) >= MAX_CART_ITEMS:
            raise CartItemLimitExceededError(MAX_CART_ITEMS)

        # Enforce: no duplicates — update quantity instead
        existing = self._find_item(book_id)
        if existing:
            raise BookAlreadyInCartError(str(book_id))

        item = CartItem(
            book_id=book_id,
            book_title=book_title,
            unit_price=unit_price,
            quantity=quantity,
        )
        self._items.append(item)
        self._record(BookAddedToCart(
            cart_id=self.id,
            customer_id=self.customer_id.value,
            book_id=book_id.value,
            book_title=book_title,
            unit_price=unit_price,
            quantity=quantity,
        ))

    def remove_book(self, book_id: BookId) -> None:
        item = self._find_item(book_id)
        if item:
            self._items.remove(item)
            self._record(BookRemovedFromCart(
                cart_id=self.id,
                book_id=book_id.value,
            ))

    def apply_coupon(self, coupon: Coupon) -> None:
        if self._coupon is not None:
            raise CouponAlreadyAppliedError(
                f"Coupon {self._coupon.code} is already applied. "
                f"Remove it before applying a new one."
            )
        self._coupon = coupon
        self._record(CouponApplied(
            cart_id=self.id,
            coupon_code=coupon.code,
            discount_percent=coupon.discount_percent,
        ))

    def checkout(self) -> None:
        if not self._items:
            raise EmptyCartError("Cannot check out an empty cart")
        self._record(CartCheckedOut(
            cart_id=self.id,
            customer_id=self.customer_id.value,
            total_amount=self.total,
        ))

    # ── Queries ────────────────────────────────────────────────────
    @property
    def items(self) -> list[CartItem]:
        return list(self._items)   # defensive copy

    @property
    def subtotal(self) -> Money:
        if not self._items:
            return Money.zero()
        result = self._items[0].subtotal
        for item in self._items[1:]:
            result = result.add(item.subtotal)
        return result

    @property
    def total(self) -> Money:
        sub = self.subtotal
        if self._coupon:
            return self._coupon.apply_to(sub)
        return sub

    @property
    def item_count(self) -> int:
        return sum(item.quantity for item in self._items)

    def is_empty(self) -> bool:
        return len(self._items) == 0

    # ── Event infrastructure ───────────────────────────────────────
    def pull_events(self) -> list[DomainEvent]:
        events = list(self._events)
        self._events.clear()
        return events

    # ── Private helpers ────────────────────────────────────────────
    def _find_item(self, book_id: BookId) -> Optional[CartItem]:
        return next((i for i in self._items if i.book_id == book_id), None)

    def _record(self, event: DomainEvent) -> None:
        self._events.append(event)

The Order Aggregate¶

# domain/order.py
from dataclasses import dataclass, field
from datetime import datetime, UTC, timedelta
from decimal import Decimal
from enum import Enum
from uuid import UUID, uuid4
from typing import Optional

from .value_objects import Money, BookId, CustomerId, Address
from .events import (
    OrderPlaced, OrderConfirmed, OrderCancelled,
    ReturnRequested, DomainEvent
)
from .exceptions import (
    InvalidOrderStatusTransitionError, OrderAlreadyConfirmedError,
    ReturnWindowExpiredError
)
from .cart import Cart, CartItem

RETURN_WINDOW_DAYS = 30


class OrderStatus(Enum):
    PENDING_PAYMENT = "pending_payment"
    CONFIRMED = "confirmed"
    PROCESSING = "processing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"
    RETURN_REQUESTED = "return_requested"
    REFUNDED = "refunded"

    def can_cancel(self) -> bool:
        return self in (
            OrderStatus.PENDING_PAYMENT,
            OrderStatus.CONFIRMED,
            OrderStatus.PROCESSING,
        )

    def can_request_return(self) -> bool:
        return self in (OrderStatus.DELIVERED,)


@dataclass(frozen=True)
class OrderLineItem:
    """
    Value Object — a frozen snapshot of a cart item.
    Prices and titles do NOT update when the catalog changes.
    This is intentional — it records what the customer agreed to pay.
    """
    book_id: BookId
    book_title: str        # snapshot — not a live reference
    unit_price: Money      # snapshot — price at time of order
    quantity: int

    @property
    def subtotal(self) -> Money:
        return self.unit_price.multiply(Decimal(self.quantity))


@dataclass
class Order:
    """
    Aggregate Root for a customer's purchase.

    Invariants enforced:
    - Status can only follow valid transitions
    - Cannot cancel a shipped/delivered order
    - Returns only within 30-day window
    - Line items are immutable after order placement
    """
    id: UUID
    customer_id: CustomerId
    cart_id: UUID
    status: OrderStatus
    shipping_address: Address
    _line_items: list[OrderLineItem]
    placed_at: datetime
    confirmed_at: Optional[datetime] = None
    delivered_at: Optional[datetime] = None
    _events: list[DomainEvent] = field(default_factory=list)

    # ── Factory ────────────────────────────────────────────────────
    @classmethod
    def place_from_cart(
        cls,
        cart: Cart,
        shipping_address: Address,
    ) -> 'Order':
        """
        Creates an Order from a Cart.
        Takes a snapshot of cart items — Cart and Order then evolve independently.
        """
        if cart.is_empty():
            raise ValueError("Cannot place an order from an empty cart")

        # Snapshot all cart items into OrderLineItems
        line_items = [
            OrderLineItem(
                book_id=item.book_id,
                book_title=item.book_title,
                unit_price=item.unit_price,
                quantity=item.quantity,
            )
            for item in cart.items
        ]

        order = cls(
            id=uuid4(),
            customer_id=cart.customer_id,
            cart_id=cart.id,
            status=OrderStatus.PENDING_PAYMENT,
            shipping_address=shipping_address,
            _line_items=line_items,
            placed_at=datetime.now(UTC),
        )
        order._record(OrderPlaced(
            order_id=order.id,
            customer_id=cart.customer_id.value,
            cart_id=cart.id,
            total_amount=order.total,
            item_count=len(line_items),
        ))
        return order

    # ── Commands ───────────────────────────────────────────────────
    def confirm(self) -> None:
        if self.status != OrderStatus.PENDING_PAYMENT:
            raise InvalidOrderStatusTransitionError(
                self.status.value, OrderStatus.CONFIRMED.value
            )
        self.status = OrderStatus.CONFIRMED
        self.confirmed_at = datetime.now(UTC)
        self._record(OrderConfirmed(
            order_id=self.id,
            customer_id=self.customer_id.value,
            total_amount=self.total,
        ))

    def cancel(self, reason: str) -> None:
        if not self.status.can_cancel():
            raise InvalidOrderStatusTransitionError(
                self.status.value, OrderStatus.CANCELLED.value
            )
        self.status = OrderStatus.CANCELLED
        self._record(OrderCancelled(
            order_id=self.id,
            customer_id=self.customer_id.value,
            reason=reason,
            refund_amount=self.total,
        ))

    def mark_shipped(self) -> None:
        if self.status != OrderStatus.PROCESSING:
            raise InvalidOrderStatusTransitionError(
                self.status.value, OrderStatus.SHIPPED.value
            )
        self.status = OrderStatus.SHIPPED

    def mark_delivered(self) -> None:
        if self.status != OrderStatus.SHIPPED:
            raise InvalidOrderStatusTransitionError(
                self.status.value, OrderStatus.DELIVERED.value
            )
        self.status = OrderStatus.DELIVERED
        self.delivered_at = datetime.now(UTC)

    def request_return(self, reason: str) -> None:
        if not self.status.can_request_return():
            raise InvalidOrderStatusTransitionError(
                self.status.value, OrderStatus.RETURN_REQUESTED.value
            )
        if not self._within_return_window():
            raise ReturnWindowExpiredError(
                f"Return window of {RETURN_WINDOW_DAYS} days has expired"
            )
        self.status = OrderStatus.RETURN_REQUESTED
        self._record(ReturnRequested(
            order_id=self.id,
            customer_id=self.customer_id.value,
            reason=reason,
        ))

    # ── Queries ────────────────────────────────────────────────────
    @property
    def line_items(self) -> list[OrderLineItem]:
        return list(self._line_items)   # defensive copy

    @property
    def total(self) -> Money:
        if not self._line_items:
            return Money.zero()
        result = self._line_items[0].subtotal
        for item in self._line_items[1:]:
            result = result.add(item.subtotal)
        return result

    def _within_return_window(self) -> bool:
        if not self.delivered_at:
            return False
        window_end = self.delivered_at + timedelta(days=RETURN_WINDOW_DAYS)
        return datetime.now(UTC) <= window_end

    # ── Event infrastructure ───────────────────────────────────────
    def pull_events(self) -> list[DomainEvent]:
        events = list(self._events)
        self._events.clear()
        return events

    def _record(self, event: DomainEvent) -> None:
        self._events.append(event)

Repositories¶

The interface lives in the domain layer; the implementation lives in infrastructure:

# domain/repositories.py (interface)
from abc import ABC, abstractmethod
from uuid import UUID
from typing import Optional
from .cart import Cart
from .order import Order
from .value_objects import CustomerId


class CartRepository(ABC):
    @abstractmethod
    def save(self, cart: Cart) -> None: ...

    @abstractmethod
    def find_by_id(self, cart_id: UUID) -> Optional[Cart]: ...

    @abstractmethod
    def find_active_by_customer(self, customer_id: CustomerId) -> Optional[Cart]: ...


class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> None: ...

    @abstractmethod
    def find_by_id(self, order_id: UUID) -> Optional[Order]: ...

    @abstractmethod
    def find_by_customer(self, customer_id: CustomerId) -> list[Order]: ...

Application Services (Use Cases)¶

Application services orchestrate without containing business logic:

# application/order_service.py
from uuid import UUID
from dataclasses import dataclass

from ..domain.cart import Cart
from ..domain.order import Order
from ..domain.value_objects import CustomerId, BookId, Money, Address, Coupon
from ..domain.repositories import CartRepository, OrderRepository
from ..domain.exceptions import DomainError


@dataclass
class AddBookToCartCommand:
    customer_id: UUID
    book_id: UUID
    book_title: str
    unit_price_amount: str
    currency: str
    quantity: int = 1


@dataclass
class PlaceOrderCommand:
    customer_id: UUID
    street_line_1: str
    city: str
    state_or_province: str
    postal_code: str
    country_code: str


class CartApplicationService:
    def __init__(
        self,
        cart_repo: CartRepository,
        order_repo: OrderRepository,
        event_publisher,           # injected from outside
    ):
        self._carts = cart_repo
        self._orders = order_repo
        self._events = event_publisher

    def add_book_to_cart(self, cmd: AddBookToCartCommand) -> UUID:
        customer_id = CustomerId(cmd.customer_id)

        # Get or create cart
        cart = self._carts.find_active_by_customer(customer_id)
        if not cart:
            cart = Cart.create(customer_id=customer_id)

        # Delegate to domain — all business logic is there
        cart.add_book(
            book_id=BookId(cmd.book_id),
            book_title=cmd.book_title,
            unit_price=Money(amount=Decimal(cmd.unit_price_amount), currency=cmd.currency),
            quantity=cmd.quantity,
        )

        # Persist and publish events
        self._carts.save(cart)
        for event in cart.pull_events():
            self._events.publish(event)

        return cart.id

    def place_order(self, cmd: PlaceOrderCommand) -> UUID:
        customer_id = CustomerId(cmd.customer_id)

        cart = self._carts.find_active_by_customer(customer_id)
        if not cart:
            raise DomainError("No active cart found for this customer")

        address = Address(
            street_line_1=cmd.street_line_1,
            street_line_2="",
            city=cmd.city,
            state_or_province=cmd.state_or_province,
            postal_code=cmd.postal_code,
            country_code=cmd.country_code,
        )

        # Cart checks out
        cart.checkout()

        # Order is created from cart snapshot
        order = Order.place_from_cart(cart=cart, shipping_address=address)

        # Save both
        self._carts.save(cart)
        self._orders.save(order)

        # Publish all events (CartCheckedOut, OrderPlaced)
        for event in cart.pull_events():
            self._events.publish(event)
        for event in order.pull_events():
            self._events.publish(event)

        return order.id

Phase 5: Integrating Bounded Contexts with Domain Events¶

When OrderPlaced is published, three other bounded contexts need to react:

  OrderPlaced event published
        │
        ├──→ INVENTORY subscribes: "Reserve the books"
        ├──→ PAYMENTS subscribes: "Initiate payment collection"
        └──→ MEMBERSHIP subscribes: "Start earning points"

This is asynchronous event-driven integration — Orders has no compile-time dependency on Inventory, Payments, or Membership. Each context subscribes to events it cares about.

Event Schema (the Published Language)¶

Events crossing context boundaries need a stable, versioned schema:

# This is the Published Language — shared schema between contexts
# Published as JSON on a message bus (Kafka, RabbitMQ, etc.)

# orders/events/order_placed_v1.py
@dataclass(frozen=True)
class OrderPlacedV1:
    """
    Published event: emitted when a customer places an order.
    Consumers: Inventory, Payments, Membership, Notifications.
    Schema version: 1 (additive changes only — never remove fields).
    """
    schema_version: str = "1"
    event_type: str = "order.placed"

    order_id: str
    customer_id: str
    placed_at: str          # ISO 8601
    total_amount: str       # decimal string to avoid float issues
    currency: str
    item_count: int
    items: list[dict]       # [{book_id, title, quantity, unit_price}]

    def to_dict(self) -> dict:
        return {
            "schema_version": self.schema_version,
            "event_type": self.event_type,
            "order_id": self.order_id,
            "customer_id": self.customer_id,
            "placed_at": self.placed_at,
            "total_amount": self.total_amount,
            "currency": self.currency,
            "item_count": self.item_count,
            "items": self.items,
        }

The Inventory Context Subscribes¶

# inventory/application/event_handlers.py

class OrderPlacedHandler:
    """
    Inventory reacts to orders placed in the Orders context.
    This is the Anti-Corruption Layer — we translate the external event
    into our own domain language before processing.
    """
    def __init__(self, stock_service: StockReservationService):
        self._stock = stock_service

    def handle(self, event_data: dict) -> None:
        # Translate external schema → our domain concepts (ACL)
        order_id = event_data["order_id"]
        items = [
            StockReservationItem(
                sku=item["book_id"],         # "book_id" in Orders = "sku" in Inventory
                quantity=item["quantity"],
            )
            for item in event_data["items"]
        ]

        # Delegate to our own domain logic
        self._stock.reserve_for_order(order_id=order_id, items=items)

Handling Eventual Consistency¶

When you use async events, you must handle scenarios where one step succeeds and another fails:

  Scenario: OrderPlaced published, but Inventory reservation fails

  Option 1 — Saga / Process Manager:
  OrderPlaced → InventoryReserved → PaymentRequested
     ↑                ↓ (fails)
     └── OrderCancelled ← InventoryReservationFailed

  Option 2 — Outbox Pattern:
  Write Order + Event to DB in one transaction.
  Separate process reads Outbox table and publishes events.
  Guarantees "at least once" delivery.

# The Outbox Pattern — guarantees no event is lost on publish failure

# In your repository save():
def save(self, order: Order) -> None:
    with self._session.begin():
        # 1. Save the order record
        order_record = self._mapper.to_record(order)
        self._session.merge(order_record)

        # 2. Save events to outbox table in THE SAME TRANSACTION
        for event in order.pull_events():
            outbox_entry = OutboxEntry(
                id=uuid4(),
                event_type=type(event).__name__,
                payload=json.dumps(event.to_dict()),
                created_at=datetime.now(UTC),
                published=False,
            )
            self._session.add(outbox_entry)

# Separate background process:
# SELECT * FROM outbox WHERE published = false
# → publish each to message bus
# → mark as published
# Retry failed publishes automatically

Phase 6: Testing Your Domain Model¶

One of the best benefits of a proper domain model: your business logic is fully testable without databases, web frameworks, or external services.

Unit Testing Aggregates¶

# tests/domain/test_cart.py
import pytest
from decimal import Decimal
from uuid import uuid4

from orders.domain.cart import Cart
from orders.domain.value_objects import BookId, CustomerId, Money, Coupon
from orders.domain.exceptions import (
    BookAlreadyInCartError, CartItemLimitExceededError,
    CouponAlreadyAppliedError, EmptyCartError
)
from orders.domain.events import BookAddedToCart, CartCheckedOut


class TestCart:

    def setup_method(self):
        self.customer_id = CustomerId(uuid4())
        self.cart = Cart.create(customer_id=self.customer_id)
        self.book_id = BookId(uuid4())
        self.price = Money(Decimal("29.99"), "USD")

    def test_add_book_creates_cart_item(self):
        self.cart.add_book(self.book_id, "Clean Code", self.price, quantity=1)

        assert len(self.cart.items) == 1
        assert self.cart.items[0].book_title == "Clean Code"
        assert self.cart.total == self.price

    def test_add_same_book_twice_raises_error(self):
        self.cart.add_book(self.book_id, "Clean Code", self.price)

        with pytest.raises(BookAlreadyInCartError):
            self.cart.add_book(self.book_id, "Clean Code", self.price)

    def test_total_includes_all_items(self):
        book_2 = BookId(uuid4())
        price_2 = Money(Decimal("19.99"), "USD")

        self.cart.add_book(self.book_id, "Clean Code", self.price)
        self.cart.add_book(book_2, "The Pragmatic Programmer", price_2)

        assert self.cart.total == Money(Decimal("49.98"), "USD")

    def test_coupon_reduces_total(self):
        self.cart.add_book(self.book_id, "Clean Code", Money(Decimal("100.00"), "USD"))
        self.cart.apply_coupon(Coupon(code="SAVE20", discount_percent=20))

        assert self.cart.total == Money(Decimal("80.00"), "USD")

    def test_cannot_apply_two_coupons(self):
        self.cart.add_book(self.book_id, "Clean Code", self.price)
        self.cart.apply_coupon(Coupon(code="FIRST", discount_percent=10))

        with pytest.raises(CouponAlreadyAppliedError):
            self.cart.apply_coupon(Coupon(code="SECOND", discount_percent=20))

    def test_checkout_empty_cart_raises_error(self):
        with pytest.raises(EmptyCartError):
            self.cart.checkout()

    def test_add_book_emits_event(self):
        self.cart.add_book(self.book_id, "Clean Code", self.price)

        events = self.cart.pull_events()
        assert len(events) == 1
        assert isinstance(events[0], BookAddedToCart)
        assert events[0].book_title == "Clean Code"

    def test_checkout_emits_event_with_correct_total(self):
        self.cart.add_book(self.book_id, "Clean Code", self.price)
        self.cart.checkout()

        events = self.cart.pull_events()
        checkout_events = [e for e in events if isinstance(e, CartCheckedOut)]
        assert len(checkout_events) == 1
        assert checkout_events[0].total_amount == self.price


class TestOrder:

    def test_cannot_cancel_shipped_order(self):
        order = self._make_order_in_status(OrderStatus.SHIPPED)

        with pytest.raises(InvalidOrderStatusTransitionError):
            order.cancel("Changed my mind")

    def test_return_outside_window_raises_error(self):
        order = self._make_delivered_order(days_ago=31)

        with pytest.raises(ReturnWindowExpiredError):
            order.request_return("Book was damaged")

    def test_return_within_window_succeeds(self):
        order = self._make_delivered_order(days_ago=5)
        order.request_return("Book was damaged")

        assert order.status == OrderStatus.RETURN_REQUESTED

    def test_confirm_publishes_event(self):
        order = self._make_order_in_status(OrderStatus.PENDING_PAYMENT)
        order.confirm()

        events = order.pull_events()
        confirmed = [e for e in events if isinstance(e, OrderConfirmed)]
        assert len(confirmed) == 1
        assert confirmed[0].order_id == order.id

Integration Testing with an In-Memory Repository¶

# tests/application/test_cart_service.py

class InMemoryCartRepository(CartRepository):
    def __init__(self):
        self._store: dict[UUID, Cart] = {}

    def save(self, cart: Cart) -> None:
        self._store[cart.id] = cart

    def find_by_id(self, cart_id: UUID) -> Optional[Cart]:
        return self._store.get(cart_id)

    def find_active_by_customer(self, customer_id: CustomerId) -> Optional[Cart]:
        return next(
            (c for c in self._store.values()
             if c.customer_id == customer_id),
            None
        )


class FakeEventPublisher:
    def __init__(self):
        self.published: list = []

    def publish(self, event) -> None:
        self.published.append(event)


class TestCartApplicationService:

    def setup_method(self):
        self.cart_repo = InMemoryCartRepository()
        self.order_repo = InMemoryOrderRepository()
        self.events = FakeEventPublisher()
        self.service = CartApplicationService(
            cart_repo=self.cart_repo,
            order_repo=self.order_repo,
            event_publisher=self.events,
        )

    def test_add_book_creates_new_cart(self):
        customer_id = uuid4()
        cmd = AddBookToCartCommand(
            customer_id=customer_id,
            book_id=uuid4(),
            book_title="Domain-Driven Design",
            unit_price_amount="54.99",
            currency="USD",
        )

        cart_id = self.service.add_book_to_cart(cmd)

        cart = self.cart_repo.find_by_id(cart_id)
        assert cart is not None
        assert len(cart.items) == 1
        assert cart.items[0].book_title == "Domain-Driven Design"

    def test_add_book_publishes_event(self):
        cmd = AddBookToCartCommand(
            customer_id=uuid4(),
            book_id=uuid4(),
            book_title="Clean Architecture",
            unit_price_amount="39.99",
            currency="USD",
        )
        self.service.add_book_to_cart(cmd)

        assert len(self.events.published) == 1
        assert isinstance(self.events.published[0], BookAddedToCart)

The Swift Method: From Event Storm to Stories¶

For teams that use Jira or Linear for story management, the Swift Method (from the original blog post) provides a bridge from Event Storming output to development stories.

BORIS — Business Object Relationships and Interaction Specification¶

After Event Storming, BORIS maps which actors interact with which aggregates and what events result:

  BORIS table for Bookly Orders context:

  ┌──────────────┬─────────────────────┬──────────────────────────┐
  │ Actor        │ Command             │ Event(s)                 │
  ├──────────────┼─────────────────────┼──────────────────────────┤
  │ Customer     │ Add Book to Cart    │ BookAddedToCart          │
  │ Customer     │ Apply Coupon        │ CouponApplied            │
  │ Customer     │ Initiate Checkout   │ CartCheckedOut           │
  │ Customer     │ Place Order         │ OrderPlaced              │
  │ Customer     │ Cancel Order        │ OrderCancelled           │
  │ Customer     │ Request Return      │ ReturnRequested          │
  │ Payment GW   │ Confirm Payment     │ OrderConfirmed           │
  │ Payment GW   │ Reject Payment      │ OrderCancelled           │
  │ Warehouse    │ Dispatch Order      │ OrderShipped             │
  │ Courier      │ Confirm Delivery    │ OrderDelivered           │
  └──────────────┴─────────────────────┴──────────────────────────┘

SNAP — Specification by NAmePart¶

SNAP translates each row of BORIS into a story specification. For each Command → Event:

  SNAP for "Add Book to Cart":

  GIVEN: Customer [X] has an active cart (or will get a new one)
  WHEN:  Customer adds Book [Y] with quantity [Z]
  THEN:  Book is in the cart with correct subtotal
  AND:   BookAddedToCart event is published
  AND:   Cart total is recalculated

  Acceptance Criteria:
  ✅ Adding the same book twice raises an error (not duplicated)
  ✅ Adding more than 50 books raises an error
  ✅ Total correctly reflects the coupon discount if applied
  ✅ Checkout with an empty cart is rejected

These SNAP specs translate directly into Jira user stories and unit test cases — the spec IS the test.

Common DDD Pitfalls in Practice¶

1. Starting with the Database Schema¶

The mistake: Open a SQL editor, design the tables, then write code to populate them.

Why it fails: You end up with an anemic domain model that mirrors your tables, not your business. Business rules get scattered into stored procedures, services, and controllers.

The fix: Design the domain model first. Let the database schema be derived from the model, not the other way around.

2. Aggregate Too Large → Locking Hell¶

The symptom: Two users can't use the system simultaneously. Timeouts on basic operations.

The cause: All business objects bundled into one aggregate. Every operation locks the entire thing.

The fix: Ask "what is the minimum set of data that must be consistent?" Design the aggregate around that — and use eventual consistency (domain events) for everything else.

The mistake: The Orders service and the Inventory service both SELECT and UPDATE from the same orders table.

Why it fails: Any schema change breaks both services. You can't evolve the contexts independently. Deployment becomes coupled.

The fix: Each bounded context owns its own database. If Inventory needs order data, it subscribes to the OrderPlaced event and maintains its own projection of that data.

# BAD: Inventory queries the Orders database directly
def reserve_stock(order_id):
    order = orders_db.execute("SELECT * FROM orders WHERE id = ?", order_id)
    # Now Inventory is coupled to Orders' schema forever

# GOOD: Inventory maintains its own read model, populated by events
class InventoryOrderProjection:
    def on_order_placed(self, event: dict) -> None:
        # Store only what Inventory cares about
        self._db.execute(
            "INSERT INTO inventory_orders (order_id, items) VALUES (?, ?)",
            event["order_id"], json.dumps(event["items"])
        )

4. Putting Business Logic in Application Services¶

The symptom: Application service methods are hundreds of lines long. Your domain objects are just data classes.

The fix: Move every if statement that represents a business rule into the aggregate. Application services should read like a summary: "get this, call that, save, publish."

5. Skipping Event Storming ("We already know the domain")¶

Why this fails: You don't know what you don't know. The most valuable output of Event Storming is the surprises — the business rules that are in someone's head but not in any document.

Even if you think you know the domain, run a 2-hour event storming session. The "Wait, WHAT?" moments will pay for the time investment many times over.

DDD in Action: Your Playbook¶

Use this checklist when starting a DDD project:

Discovery Phase

Identify all stakeholders — who holds domain knowledge?
Run Event Storming (minimum 4 hours for a meaningful domain)
Build the event timeline and identify hotspots (areas of confusion/complexity)
Write the first draft of your Ubiquitous Language glossary

Design Phase

Draw candidate Bounded Context boundaries on the Event Storm output
Name each bounded context (one team = one context as a starting point)
Document the Context Map (what relationships exist, which patterns apply)
Identify aggregates per context — start small, split later if needed
Validate aggregate boundaries: "What must be consistent together?"

Implementation Phase

Create the folder structure (domain / application / infrastructure / api)
Implement Value Objects first (they're pure, easy to test)
Implement Aggregates with business invariants and domain events
Define Repository interfaces in the domain layer
Implement Application Services (thin orchestration only)
Implement Repository in the infrastructure layer
Wire event publishing through the Outbox Pattern

Validation Phase

Unit test every aggregate in isolation (no database required)
Integration test application services with in-memory repositories
Test event publishing and consumption across context boundaries
Review: does the code read like the business? Would a domain expert understand it?
Update the Ubiquitous Language glossary as understanding evolves

Summary¶

DDD in action follows a clear loop:

  Discover the domain → Draw the boundaries → Design the aggregates
         ↑                                              ↓
  Refine the model  ← Test rigorously     ← Implement in code

The key insight is that code is a model of your understanding of the domain. As your understanding deepens — through Event Storming, stakeholder conversations, and shipping features — your model should deepen with it.

The practices that make this work:

Event Storming to build shared understanding before a line of code is written
Bounded Contexts to give teams autonomy and prevent semantic collisions
Small Aggregates to enforce consistency without locking the world
Domain Events to integrate without coupling
Rich Domain Models to keep business logic where it belongs — in the domain
The Outbox Pattern to guarantee events are never silently lost
Tests that read like specifications — the SNAP format gives you both

DDD is not about patterns for their own sake. It's about software that is honest — honest about what the business does, honest about who owns what, honest about where the complexity lives. Code that any domain expert can look at and say: "Yes, that's what we do."

Want to dive deeper into the foundational theory behind these practices? Read Domain-Driven Design: The Complete Guide. Have questions? Connect on LinkedIn, X or reach out via email.

Discussion

Have thoughts on this post? Share them below — questions, corrections, or your own experience are all welcome.