paulpas Tactical Domain-Driven Design — Entities, Value Objects & Aggregate Roots
Implements the tactical building blocks of Domain-Driven Design in Python: immutable value objects with semantic equality, entities with identity and lifecycle, aggregates that enforce invariants as a unit, repository interfaces that abstract persistence, domain events for expressing state changes, and factories that encapsulate complex object creation.
TL;DR Checklist
- Define Value Objects as frozen dataclasses with semantic equality
- Define Entities with identity and protected mutation methods
- Enforce all invariants inside the Aggregate Root — never let child objects violate them
- Repository interfaces declare query methods, not CRUD operations
- Domain events are immutable snapshots published after state changes
- Factories encapsulate creation logic that cannot fit in
__init__
When to Use
- Implementing a domain model where business rules require strict invariant enforcement
- Designing objects where identity matters (Entities) versus where value matters (Value Objects)
- Building aggregates that must maintain consistency across multiple related objects
- Defining repository contracts for persistence without leaking infrastructure concerns
- Expressing state changes as explicit Domain Events for eventual consistency and audit trails
- Encapsulating complex object creation logic that would clutter constructors
When NOT to Use
- Simple DTOs or data transfer objects with no business logic — use plain Pydantic models or
TypedDict - Read-only projections used only for reporting — these do not need aggregate boundaries
- Performance-critical hot paths where the overhead of invariant checking is unacceptable (but consider adding it in non-production builds)
- When your team lacks domain experts to validate invariants — tactical DDD without a well-understood domain produces elaborate abstractions with no business value
Core Workflow
Classify Domain Objects — For each concept in the domain, decide whether it is an Entity (identity-based, stateful) or a Value Object (value-based, immutable). Checkpoint: If two objects with the same data should be interchangeable without tracking them separately, it is a Value Object. If you need to distinguish "Order #42" from "Order #43" even if both have identical line items, it is an Entity.
Define Value Objects — Create frozen
dataclasseswith semantic equality (__eq__) and hashability. Every field that contributes to the object's identity must be included in__eq__. Checkpoint: Frozen value objects cannot be mutated after creation — any "update" returns a new instance.Define Entities — Create classes with a unique identifier (UUID, integer sequence, or domain-specific key). Encapsulate state mutation behind named methods that enforce invariants before modifying state. Checkpoint: Every public mutation method must validate preconditions and raise an exception if the invariant is violated — never silently accept invalid state.
Define Aggregates — Group related entities and value objects into an Aggregate Root. Only the root's methods can modify internal state. Child objects are accessed through the root and cannot be mutated directly from outside. Checkpoint: The aggregate boundary is the consistency boundary — all invariants within the group must hold simultaneously after any operation.
Define Repository Interfaces — Create protocol classes that declare query and persistence methods. The interface lives in the domain layer; implementations live in the infrastructure layer. Checkpoint: Repository methods should reflect how the domain queries for objects, not how the database indexes them.
Implement Domain Events — Define immutable event dataclasses with a timestamp and all state needed by handlers. Publish events through the Aggregate Root after a state change is committed. Checkpoint: Events must be idempotent — handlers should guard against processing the same event twice using an event store that tracks published event IDs.
Implement Factories — Move complex object creation out of constructors into factory methods or dedicated factory classes when the logic cannot fit in a single
__init__call. Checkpoint: A factory is only needed when creation requires knowledge of multiple aggregates, external services, or multi-step validation that does not belong to any single entity's responsibilities.
Implementation Patterns
Pattern 1: Value Objects — Immutability and Semantic Equality (BAD vs. GOOD)
❌ BAD — Mutable value objects break the fundamental contract that "same value equals same object" because their identity can drift.
# ❌ BAD: A mutable Money type allows accidental modification
class BadMoney:
"""Mutable money — changing amount after comparison breaks equality."""
def __init__(self, amount: float, currency: str = "USD") -> None:
self.amount = amount # ← mutable, can be changed externally
self.currency = currency # ← also mutable
def __eq__(self, other: object) -> bool:
if not isinstance(other, BadMoney):
return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def add(self, other: "BadMoney") -> "BadMoney":
# Returns a new instance, but the original can still be mutated by callers
return BadMoney(self.amount + other.amount, self.currency)
# Usage demonstrates the problem:
m1 = BadMoney(100.0, "USD")
m2 = BadMoney(100.0, "USD")
assert m1 == m2 # True — equal values
m1.amount = 99.0 # ← silently changes m1; now they differ even though no one called add()
assert m1 != m2 # True — but m1's identity has been corrupted without its knowledge
✅ GOOD — Frozen dataclasses with semantic equality guarantee immutability and consistent hashing.
from __future__ import annotations
from dataclasses import dataclass, field
from decimal import Decimal
from typing import ClassVar
@dataclass(frozen=True, order=False)
class Money:
"""Immutable monetary value with currency.
Uses Decimal for exact arithmetic — never float for money.
Equality is based on the pair (amount, currency), not object identity.
Hashable so it can be used as a dict key or in sets.
"""
amount: Decimal
currency: str = field(default="USD", hash=True, compare=True)
# Canonical currencies — prevents typos from creating invalid states
_VALID_CURRENCIES: ClassVar[frozenset[str]] = frozenset({
"USD", "EUR", "GBP", "JPY", "CAD", "AUD", "CHF", "CNY",
})
def __post_init__(self) -> None:
if self.currency not in self._VALID_CURRENCIES:
raise ValueError(
f"Invalid currency '{self.currency}' — must be one of "
f"{sorted(self._VALID_CURRENCIES)}"
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Money):
return NotImplemented
# Currency codes are case-insensitive for comparison (ISO 4217 allows both)
return self.amount == other.amount and self.currency.upper() == other.currency.upper()
def __hash__(self) -> int:
return hash((self.amount, self.currency.upper()))
def add(self, other: Money) -> Money:
"""Add two Money values — currencies must match."""
if self.currency != other.currency:
raise ValueError(
f"Cannot add {other.currency} to {self.currency} — "
"currencies must match for arithmetic operations"
)
return Money(self.amount + other.amount, self.currency)
def subtract(self, other: Money) -> Money:
"""Subtract another Money value from this one."""
if self.currency != other.currency:
raise ValueError(
f"Cannot subtract {other.currency} from {self.currency}"
)
return Money(self.amount - other.amount, self.currency)
def multiply(self, factor: Decimal) -> Money:
"""Multiply by a scalar factor."""
if not isinstance(factor, Decimal):
raise TypeError("Factor must be a Decimal")
return Money(self.amount * factor, self.currency)
@classmethod
def from_cents(cls, cents: int, currency: str = "USD") -> Money:
"""Factory method: create Money from integer cents to avoid float entirely."""
return cls(Decimal(cents) / Decimal(100), currency)
Pattern 2: Aggregate Root — Invariant Enforcement
❌ BAD — Child objects can be mutated directly, allowing invariant violations.
# ❌ BAD: OrderItem can be added to order's items list directly
class BadOrder:
def __init__(self, customer_id: str) -> None:
self.customer_id = customer_id
self.items: list[BadOrderItem] = [] # ← caller can mutate this directly
self.total = Decimal("0.00")
def add_item(self, item: BadOrderItem) -> None:
"""Adding an item without enforcing any invariants."""
self.items.append(item) # No validation — zero quantity slips through
class BadOrderItem:
def __init__(self, product_id: str, quantity: int, unit_price: Decimal) -> None:
# No validation at all — negative quantities accepted
self.product_id = product_id
self.quantity = quantity # ← can be changed externally too
self.unit_price = unit_price # ← also mutable
✅ GOOD — The Aggregate Root controls all mutations through named methods with invariant checks.
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import UTC, datetime
from decimal import Decimal
from uuid import UUID, uuid4
@dataclass(frozen=True)
class LineItem:
"""Immutable line item within an Order aggregate."""
product_id: str
quantity: int
unit_price: Decimal
def __post_init__(self) -> None:
if self.quantity <= 0:
raise ValueError(f"Line item quantity must be positive, got {self.quantity}")
if self.unit_price < 0:
raise ValueError(f"Unit price must not be negative, got {self.unit_price}")
@property
def line_total(self) -> Decimal:
return self.unit_price * self.quantity
class OrderStatus:
"""Value object for order status — prevents invalid state transitions."""
DRAFT = "draft"
CONFIRMED = "confirmed"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
_TRANSITIONS: dict[str, set[str]] = {
OrderStatus.DRAFT: {OrderStatus.CONFIRMED, OrderStatus.CANCELLED},
OrderStatus.CONFIRMED: {OrderStatus.SHIPPED, OrderStatus.CANCELLED},
OrderStatus.SHIPPED: {OrderStatus.DELIVERED},
OrderStatus.DELIVERED: set(), # terminal state — no transitions allowed
OrderStatus.CANCELLED: set(), # terminal state
}
@classmethod
def is_valid_transition(cls, from_state: str, to_state: str) -> bool:
return to_state in cls._TRANSITIONS.get(from_state, set())
@dataclass
class Order:
"""Aggregate Root for the Order aggregate.
All mutations go through this class's methods. Internal state is
protected by convention (prefixed with _) and enforced by invariant checks.
Only one method modifies items directly: _add_line_item, which validates
before appending.
"""
id: UUID = field(default_factory=uuid4)
customer_id: str = field(default="")
_items: list[LineItem] = field(default_factory=list, repr=False)
_status: str = field(default=OrderStatus.DRAFT, repr=False)
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
# --- Public mutation methods — all enforce invariants ---
def set_customer(self, customer_id: str) -> None:
"""Assign a customer to this order. Only valid in DRAFT status."""
self._enforce_status(OrderStatus.DRAFT, "set_customer")
if not customer_id or len(customer_id) < 3:
raise ValueError(f"Customer ID must be at least 3 characters, got '{customer_id}'")
self.customer_id = customer_id
def add_line_item(self, product_id: str, quantity: int, unit_price: Decimal) -> None:
"""Add a line item — creates an immutable LineItem and validates the aggregate."""
self._enforce_status(OrderStatus.DRAFT, "add_line_item")
# Create the value object (it validates itself in __post_init__)
item = LineItem(product_id=product_id, quantity=quantity, unit_price=unit_price)
self._items.append(item)
self._recalculate_total()
def confirm(self) -> None:
"""Transition order to confirmed status. Requires a customer assigned."""
self._enforce_status(OrderStatus.DRAFT, "confirm")
if not self.customer_id:
raise ValueError("Cannot confirm an order without a customer assigned")
if not self._items:
raise ValueError("Cannot confirm an order with no line items")
self._status = OrderStatus.CONFIRMED
def cancel(self) -> None:
"""Cancel the order. Valid from DRAFT or CONFIRMED."""
valid_from = {OrderStatus.DRAFT, OrderStatus.CONFIRMED}
if self._status not in valid_from:
raise ValueError(
f"Cannot cancel an order in '{self._status}' status — "
f"only valid from {valid_from}"
)
self._status = OrderStatus.CANCELLED
# --- Read-only accessors (no mutation possible) ---
@property
def items(self) -> tuple[LineItem, ...]:
"""Returns a snapshot — callers cannot modify the internal list."""
return tuple(self._items)
@property
def status(self) -> str:
return self._status
@property
def total(self) -> Decimal:
"""Calculated total of all line items."""
return sum((item.line_total for item in self._items), Decimal("0.00"))
# --- Private helpers (enforce internal consistency) ---
def _enforce_status(self, required_status: str, method_name: str) -> None: # noqa: ARG002
if self._status != required_status:
raise ValueError(
f"Cannot call '{method_name}' — order is in status "
f"'{self._status}', requires '{required_status}'"
)
def _recalculate_total(self) -> None: # noqa: ARG002
"""Internal method — called after every item addition.
In a larger aggregate, this would also update cached totals,
notify internal subscribers of domain events, and revalidate
any derived invariants.
"""
pass # Total is computed on-demand; no caching needed for small orders
Pattern 3: Repository Protocol and Domain Events
✅ GOOD — Repository interfaces are pure protocols that define domain-level query semantics.
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Generic, Protocol, TypeVar
from uuid import UUID
T_co = TypeVar("T_co", covariant=True)
class RepositoryProtocol(Protocol[T_co]):
"""Base protocol for all repository interfaces.
Domain code depends on this protocol; infrastructure implements it.
Methods reflect how the domain queries, not how the database indexes.
"""
def get_by_id(self, id: UUID) -> T_co | None: ...
def save(self, entity: T_co) -> None: ...
def delete(self, entity: T_co) -> None: ...
class OrderRepositoryProtocol(RepositoryProtocol["Order"], Protocol): # type: ignore[name-defined]
"""Domain-specific repository protocol with query methods the domain needs."""
def find_by_customer_id(self, customer_id: str) -> Sequence[Order]: ...
def find_open_orders_for_date(self, date: datetime) -> Sequence[Order]: ...
# --- Domain Events ---
@dataclass(frozen=True)
class OrderCreatedEvent:
"""Immutable snapshot of the order state at creation time."""
order_id: UUID
customer_id: str
item_count: int
total_amount: Decimal
occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC))
@dataclass(frozen=True)
class OrderConfirmedEvent:
order_id: UUID
confirmed_at: datetime = field(default_factory=lambda: datetime.now(UTC))
@dataclass(frozen=True)
class OrderCancelledEvent:
order_id: UUID
cancellation_reason: str
occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC))
class DomainEventPublisher:
"""Publishes domain events with idempotency support.
In production, this would persist events to an event store and notify handlers.
Here it demonstrates the pattern of collecting events on the aggregate
and flushing them after the unit of work commits.
"""
def __init__(self) -> None:
self._published_ids: set[UUID] = set()
def publish(self, event: OrderCreatedEvent | OrderConfirmedEvent | OrderCancelledEvent) -> None:
# Idempotency guard — prevent duplicate processing
if event.occurred_at.timestamp() in self._published_ids:
return # Already published; skip silently
self._published_ids.add(event.occurred_at.timestamp())
self._dispatch(event)
def _dispatch(self, event: OrderCreatedEvent | OrderConfirmedEvent | OrderCancelledEvent) -> None:
"""In production, this routes to registered handlers."""
pass # Event routing handled by infrastructure layer
class InMemoryOrderRepository:
"""Concrete implementation for demonstration. In production, use SQLAlchemy, DDD-style repository, etc."""
def __init__(self) -> None:
self._store: dict[UUID, Order] = {}
def get_by_id(self, id: UUID) -> Order | None:
return self._store.get(id)
def save(self, entity: Order) -> None:
self._store[entity.id] = entity
def delete(self, entity: Order) -> None:
self._store.pop(entity.id, None)
def find_by_customer_id(self, customer_id: str) -> Sequence[Order]:
return [
order for order in self._store.values()
if order.customer_id == customer_id
]
def find_open_orders_for_date(self, date: datetime) -> Sequence[Order]:
return [
order for order in self._store.values()
if order.status in (OrderStatus.DRAFT, OrderStatus.CONFIRMED)
]
Constraints
MUST DO
- Make Value Objects frozen dataclasses with explicit
__eq__and__hash__— they must be safe to use as dictionary keys and set members - Define a unique identifier for every Entity — even if it is a natural key (e.g., ISO currency code), express it explicitly in the type
- Enforce all domain invariants inside the Aggregate Root's mutation methods — never allow an aggregate to exist in an invalid state, even temporarily
- Declare repository interfaces as
Protocolclasses in the domain layer — implementations must be in the infrastructure layer, never imported by domain code - Publish domain events as frozen dataclasses that capture a complete snapshot of the relevant state at the moment the event occurred
- Use factory methods or dedicated factory classes when object creation requires knowledge of multiple aggregates, external services, or multi-step validation
- Use
Decimal(notfloat) for all monetary values to prevent floating-point rounding errors
MUST NOT DO
- Mutate a Value Object after creation — any "update" must return a new instance; frozen dataclasses enforce this at the language level
- Allow child objects in an Aggregate to be mutated directly from outside — always route modifications through the Aggregate Root's public methods
- Design repositories around database operations (
findByCustomerId,deleteWhere); design them around domain queries (find_by_customer_id,delete) - Publish domain events before the unit of work that caused them is committed — an event should never be published for a state change that is later rolled back
- Put persistence logic inside entities or value objects — entities must know nothing about databases, ORMs, or serialization formats
- Use strings as entity IDs in production without a dedicated identifier type (e.g.,
OrderIDwrappingUUID) — this hides bugs where order IDs and customer IDs are swapped
Related Skills
| Skill | Purpose |
|---|---|
domain-driven-strategic |
Defines the bounded contexts and architecture boundaries within which these tactical patterns operate — strategic design must precede tactical implementation |
architecture-microservices |
Guides how aggregates mapped to a single bounded context become independently deployable microservices with their own data stores |
Live References
Authoritative documentation and reference material for tactical DDD patterns.
- Domain-Driven Design: Tackling Complexity in the Heart of Software by Eric Evans — The original definition of Entities, Value Objects, Aggregates, and their boundaries
- Implementing Domain-Driven Design by Vaughn Vernon ("The Red Book") — Chapter 8 covers aggregates and consistency boundaries; Chapter 11 covers repositories
- Udi Dahan — Entities, Value Objects, and Monitors — Udi's practical guidance on choosing between entity types
- Python dataclasses documentation — Reference for
@dataclass(frozen=True)and field options likehash=True - Python typing module — Protocol support (PEP 544) — Structured subtyping via Protocol classes for repository interfaces
- DDD Lite — Vaughn Vernon's simplified DDD — A pragmatic introduction for teams finding full DDD too heavy
- The CQRS Cookbook — Domain Events pattern — How domain events enable eventual consistency between read and write models