Domain-Driven Design & Hexagonal Architecture in Python¶
A Practical Primer¶
Published: February 24, 2026 · Last edited: February 24, 2026
Introduction¶
Domain-Driven Design (DDD) and Hexagonal Architecture are two complementary approaches to structuring complex software systems. DDD focuses on modeling the business domain accurately in code, while Hexagonal Architecture ensures the domain remains isolated from infrastructure concerns (databases, HTTP, messaging, etc.).
Together they produce systems that are: - Testable in isolation — no database required to test business logic - Adaptable — swap databases, frameworks, or messaging systems with minimal friction - Expressive — code reads like the business domain, not like infrastructure plumbing
Domain-Driven Design¶
Core Philosophy¶
DDD (popularized by Eric Evans in "Domain-Driven Design: Tackling Complexity in the Heart of Software", 2003) argues that the primary challenge in software is understanding the business domain, and that the solution is to build a rich domain model that mirrors how domain experts think and speak.
Key principles: - Ubiquitous Language — developers and domain experts share a single vocabulary; that vocabulary lives in the code - Model-Driven Design — the code is the model; there's no translation layer between "what the business means" and "what the code does" - Bounded Contexts — large domains are divided into smaller, coherent sub-domains with explicit boundaries
Building Blocks¶
1. Value Objects¶
A Value Object has no identity — it is defined entirely by its attributes. Two value objects with the same attributes are equal. They are immutable.
from dataclasses import dataclass
from typing import Self
@dataclass(frozen=True)
class Money:
amount: float
currency: str
def __post_init__(self) -> None:
if self.amount < 0:
raise ValueError("Amount cannot be negative")
if not self.currency:
raise ValueError("Currency is required")
def add(self, other: Self) -> Self:
if self.currency != other.currency:
raise ValueError(f"Cannot add {self.currency} and {other.currency}")
return Money(self.amount + other.amount, self.currency)
def multiply(self, factor: float) -> Self:
return Money(round(self.amount * factor, 2), self.currency)
def __str__(self) -> str:
return f"{self.amount:.2f} {self.currency}"
@dataclass(frozen=True)
class Address:
street: str
city: str
country: str
postal_code: str
# Usage
price = Money(10.00, "USD")
tax = Money(1.50, "USD")
total = price.add(tax) # Money(11.50, "USD")
# Equality by value
assert Money(10.00, "USD") == Money(10.00, "USD") # True
Rules for Value Objects:
- Always immutable (frozen=True in dataclasses)
- Validate in __post_init__
- Contain domain logic relevant to the value (e.g., add, multiply)
- Never have an id field
2. Entities¶
An Entity has a unique identity that persists over time, even if its attributes change. Two entities with the same attributes but different IDs are not equal.
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from datetime import datetime
from typing import Optional
@dataclass
class Customer:
name: str
email: str
address: Address
id: UUID = field(default_factory=uuid4)
created_at: datetime = field(default_factory=datetime.utcnow)
_loyalty_points: int = field(default=0, repr=False)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Customer):
return False
return self.id == other.id
def __hash__(self) -> int:
return hash(self.id)
# Domain behaviour lives on the entity
def add_loyalty_points(self, points: int) -> None:
if points <= 0:
raise ValueError("Points must be positive")
self._loyalty_points += points
def redeem_loyalty_points(self, points: int) -> None:
if points > self._loyalty_points:
raise ValueError("Insufficient loyalty points")
self._loyalty_points -= points
@property
def loyalty_points(self) -> int:
return self._loyalty_points
def change_address(self, new_address: Address) -> None:
# Could raise domain events here
self.address = new_address
3. Aggregates and Aggregate Roots¶
An Aggregate is a cluster of domain objects (entities + value objects) treated as a single unit for data changes. The Aggregate Root is the entry point — external code can only interact with the aggregate through the root.
Aggregates enforce invariants (business rules that must always hold).
from dataclasses import dataclass, field
from enum import Enum
from uuid import UUID, uuid4
from typing import List
from datetime import datetime
class OrderStatus(Enum):
PENDING = "pending"
CONFIRMED = "confirmed"
SHIPPED = "shipped"
DELIVERED = "delivered"
CANCELLED = "cancelled"
@dataclass
class OrderLine:
"""Entity within the Order aggregate."""
product_id: UUID
product_name: str
quantity: int
unit_price: Money
id: UUID = field(default_factory=uuid4)
def __post_init__(self) -> None:
if self.quantity <= 0:
raise ValueError("Quantity must be positive")
@property
def subtotal(self) -> Money:
return self.unit_price.multiply(self.quantity)
@dataclass
class Order:
"""Aggregate Root. All mutations go through this class."""
customer_id: UUID
shipping_address: Address
id: UUID = field(default_factory=uuid4)
status: OrderStatus = field(default=OrderStatus.PENDING)
_lines: List[OrderLine] = field(default_factory=list, repr=False)
created_at: datetime = field(default_factory=datetime.utcnow)
_domain_events: List = field(default_factory=list, repr=False)
def __eq__(self, other: object) -> bool:
if not isinstance(other, Order):
return False
return self.id == other.id
def __hash__(self) -> int:
return hash(self.id)
# --- Aggregate invariant enforcement ---
def add_line(self, product_id: UUID, product_name: str,
quantity: int, unit_price: Money) -> None:
"""Add a line item. Only allowed on PENDING orders."""
self._assert_can_modify()
# Check if the product already exists in the order
for line in self._lines:
if line.product_id == product_id:
raise ValueError(f"Product {product_id} already in order. Use update_quantity instead.")
self._lines.append(OrderLine(product_id, product_name, quantity, unit_price))
def update_quantity(self, product_id: UUID, new_quantity: int) -> None:
self._assert_can_modify()
line = self._find_line(product_id)
if new_quantity <= 0:
self._lines.remove(line)
else:
line.quantity = new_quantity
def confirm(self) -> None:
if not self._lines:
raise ValueError("Cannot confirm an empty order")
if self.status != OrderStatus.PENDING:
raise ValueError(f"Cannot confirm order in status {self.status.value}")
self.status = OrderStatus.CONFIRMED
self._record_event(OrderConfirmed(order_id=self.id, customer_id=self.customer_id))
def cancel(self) -> None:
if self.status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
raise ValueError(f"Cannot cancel order that has been {self.status.value}")
self.status = OrderStatus.CANCELLED
self._record_event(OrderCancelled(order_id=self.id))
@property
def lines(self) -> List[OrderLine]:
return list(self._lines) # Return a copy to protect internal state
@property
def total(self) -> Money:
if not self._lines:
return Money(0.00, "USD")
result = self._lines[0].subtotal
for line in self._lines[1:]:
result = result.add(line.subtotal)
return result
@property
def domain_events(self) -> list:
return list(self._domain_events)
def clear_events(self) -> None:
self._domain_events.clear()
# --- Private helpers ---
def _assert_can_modify(self) -> None:
if self.status != OrderStatus.PENDING:
raise ValueError(f"Cannot modify order in status {self.status.value}")
def _find_line(self, product_id: UUID) -> OrderLine:
for line in self._lines:
if line.product_id == product_id:
return line
raise ValueError(f"Product {product_id} not found in order")
def _record_event(self, event) -> None:
self._domain_events.append(event)
4. Domain Events¶
Domain Events record that something significant happened in the domain. They enable loose coupling between aggregates and trigger side effects (emails, projections, etc.).
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from datetime import datetime
@dataclass(frozen=True)
class DomainEvent:
"""Base class for all domain events."""
occurred_at: datetime = field(default_factory=datetime.utcnow)
event_id: UUID = field(default_factory=uuid4)
@dataclass(frozen=True)
class OrderConfirmed(DomainEvent):
order_id: UUID = field(default=None)
customer_id: UUID = field(default=None)
@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
order_id: UUID = field(default=None)
@dataclass(frozen=True)
class OrderShipped(DomainEvent):
order_id: UUID = field(default=None)
tracking_number: str = field(default=None)
5. Domain Services¶
When business logic doesn't naturally belong to a single entity or value object, it lives in a Domain Service. Domain services are stateless and operate purely on domain objects.
from decimal import Decimal
class PricingService:
"""
Domain service for calculating order totals with discounts.
This logic spans multiple aggregates and doesn't belong to one.
"""
def calculate_discounted_total(
self, order: Order, customer: Customer
) -> Money:
base_total = order.total
discount_rate = self._get_discount_rate(customer.loyalty_points)
discount = base_total.multiply(discount_rate)
return base_total.add(Money(-discount.amount, base_total.currency))
def _get_discount_rate(self, loyalty_points: int) -> float:
if loyalty_points >= 1000:
return 0.15 # 15% discount
elif loyalty_points >= 500:
return 0.10 # 10% discount
elif loyalty_points >= 100:
return 0.05 # 5% discount
return 0.0
6. Repositories¶
A Repository provides a collection-like interface for accessing aggregates. It abstracts persistence — the domain doesn't know or care whether data lives in Postgres, MongoDB, or memory.
This is where DDD meets Hexagonal Architecture: the repository interface is defined in the domain, but the implementation lives in the infrastructure layer.
from abc import ABC, abstractmethod
from typing import Optional, List
from uuid import UUID
class OrderRepository(ABC):
"""Port (interface) defined in the domain layer."""
@abstractmethod
def save(self, order: Order) -> None:
"""Persist an order (insert or update)."""
...
@abstractmethod
def find_by_id(self, order_id: UUID) -> Optional[Order]:
"""Return an order by ID, or None if not found."""
...
@abstractmethod
def find_by_customer(self, customer_id: UUID) -> List[Order]:
"""Return all orders for a customer."""
...
@abstractmethod
def delete(self, order_id: UUID) -> None:
...
class CustomerRepository(ABC):
@abstractmethod
def save(self, customer: Customer) -> None:
...
@abstractmethod
def find_by_id(self, customer_id: UUID) -> Optional[Customer]:
...
@abstractmethod
def find_by_email(self, email: str) -> Optional[Customer]:
...
Strategic Design¶
When systems grow large, DDD introduces strategic patterns to manage complexity across sub-domains.
Bounded Contexts¶
A Bounded Context is an explicit boundary within which a particular domain model applies. The same word can mean different things in different contexts.
┌─────────────────────────┐ ┌─────────────────────────┐
│ Order Context │ │ Inventory Context │
│ │ │ │
│ Customer │ │ Product │
│ - id │ │ - id │
│ - name │ │ - sku │
│ - loyaltyPoints │ │ - stockLevel │
│ │ │ - reorderThreshold │
│ Order │ │ │
│ OrderLine │ │ StockReservation │
└─────────────────────────┘ └─────────────────────────┘
│ │
└──────── Context Map ─────────────┘
Each bounded context owns its model, its database schema, and communicates with others through explicit contracts (APIs, events).
Context Map (Anti-Corruption Layer)¶
When integrating with external systems or other bounded contexts, an Anti-Corruption Layer (ACL) translates between their model and yours, protecting your domain from external pollution.
# External shipping API returns this dict structure
# We don't want that leaking into our domain
@dataclass(frozen=True)
class ExternalShipmentStatus:
"""Their model."""
shipment_id: str
current_state: str # "IN_TRANSIT", "DELIVERED", etc.
eta_timestamp: Optional[int] # Unix timestamp
class ShippingAntiCorruptionLayer:
"""Translates external shipping model into our domain concepts."""
def translate_status(self, external: ExternalShipmentStatus) -> OrderStatus:
mapping = {
"LABEL_CREATED": OrderStatus.CONFIRMED,
"IN_TRANSIT": OrderStatus.SHIPPED,
"OUT_FOR_DELIVERY": OrderStatus.SHIPPED,
"DELIVERED": OrderStatus.DELIVERED,
}
return mapping.get(external.current_state, OrderStatus.SHIPPED)
Hexagonal Architecture¶
The Core Idea¶
Hexagonal Architecture (also called Ports & Adapters), coined by Alistair Cockburn in 2005, puts the application at the center and treats everything else (HTTP, databases, message queues, CLI) as external actors that communicate through well-defined interfaces.
┌──────────────────────────────────┐
│ │
HTTP ────────►│ [Adapter] [Port] │
│ │
CLI ────────►│ [Adapter] [Port] Domain + │
│ ↕ Application │
Tests ───────►│ [Adapter] [Port] Core │
│ │
│ [Port] [Adapter] ──┼──► PostgreSQL
│ │
│ [Port] [Adapter] ──┼──► Email Service
│ │
└──────────────────────────────────┘
The hexagon shape is just a metaphor — the important thing is inside vs outside.
- Inside: Domain model + Application services (pure Python, no I/O)
- Outside: All I/O and infrastructure (databases, HTTP, email, files)
- Ports: Interfaces that define how the inside communicates with the outside
- Adapters: Concrete implementations of those interfaces
Two Types of Ports¶
| Type | Direction | Who defines it | Who calls it |
|---|---|---|---|
| Driving (Primary) | Outside → Inside | Application | External actor (HTTP, CLI) |
| Driven (Secondary) | Inside → Outside | Application | Application core |
Ports¶
Ports are Python ABCs (Abstract Base Classes) or Protocols. They represent capabilities needed by the application.
# --- Driving Port (what the application exposes) ---
from abc import ABC, abstractmethod
from dataclasses import dataclass
from uuid import UUID
from typing import List, Optional
@dataclass
class CreateOrderCommand:
customer_id: UUID
shipping_address_street: str
shipping_address_city: str
shipping_address_country: str
shipping_address_postal: str
@dataclass
class AddOrderLineCommand:
order_id: UUID
product_id: UUID
product_name: str
quantity: int
unit_price_amount: float
unit_price_currency: str
@dataclass
class OrderSummaryDTO:
"""Data Transfer Object — plain data, no domain logic."""
order_id: str
customer_id: str
status: str
total_amount: float
total_currency: str
line_count: int
class OrderApplicationPort(ABC):
"""Driving port: what the outside world can ask our application to do."""
@abstractmethod
def create_order(self, cmd: CreateOrderCommand) -> UUID:
...
@abstractmethod
def add_line_to_order(self, cmd: AddOrderLineCommand) -> None:
...
@abstractmethod
def confirm_order(self, order_id: UUID) -> None:
...
@abstractmethod
def cancel_order(self, order_id: UUID) -> None:
...
@abstractmethod
def get_order_summary(self, order_id: UUID) -> Optional[OrderSummaryDTO]:
...
# --- Driven Ports (what the application needs from the outside) ---
class EventPublisher(ABC):
"""Driven port: the application needs to publish events somewhere."""
@abstractmethod
def publish(self, event: DomainEvent) -> None:
...
class EmailNotifier(ABC):
"""Driven port: the application needs to send emails."""
@abstractmethod
def send_order_confirmation(self, customer_email: str, order_id: UUID) -> None:
...
Adapters¶
Adapters implement the ports. They're the glue between your application and the real world.
# =====================================================
# PRIMARY (DRIVING) ADAPTERS
# =====================================================
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class CreateOrderRequest(BaseModel):
customer_id: str
street: str
city: str
country: str
postal_code: str
class FastAPIOrderAdapter:
"""
HTTP adapter that drives the application via its port.
Translates HTTP concerns into application commands.
"""
def __init__(self, order_service: OrderApplicationPort) -> None:
self._service = order_service
def create_order(self, request: CreateOrderRequest) -> dict:
cmd = CreateOrderCommand(
customer_id=UUID(request.customer_id),
shipping_address_street=request.street,
shipping_address_city=request.city,
shipping_address_country=request.country,
shipping_address_postal=request.postal_code,
)
try:
order_id = self._service.create_order(cmd)
return {"order_id": str(order_id)}
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
# =====================================================
# SECONDARY (DRIVEN) ADAPTERS
# =====================================================
import json
from typing import Dict
class InMemoryOrderRepository(OrderRepository):
"""
In-memory adapter for the OrderRepository port.
Perfect for tests and development.
"""
def __init__(self) -> None:
self._store: Dict[UUID, Order] = {}
def save(self, order: Order) -> None:
self._store[order.id] = order
def find_by_id(self, order_id: UUID) -> Optional[Order]:
return self._store.get(order_id)
def find_by_customer(self, customer_id: UUID) -> List[Order]:
return [o for o in self._store.values() if o.customer_id == customer_id]
def delete(self, order_id: UUID) -> None:
self._store.pop(order_id, None)
class PostgresOrderRepository(OrderRepository):
"""
PostgreSQL adapter for the OrderRepository port.
Uses raw SQL or an ORM — the domain doesn't care.
"""
def __init__(self, connection_string: str) -> None:
# In real code: self._engine = create_engine(connection_string)
self._connection_string = connection_string
def save(self, order: Order) -> None:
# Serialize Order -> ORM model -> DB
# This is where ORM mapping lives, isolated from domain
pass
def find_by_id(self, order_id: UUID) -> Optional[Order]:
# Query DB -> ORM model -> reconstruct Order aggregate
pass
def find_by_customer(self, customer_id: UUID) -> List[Order]:
pass
def delete(self, order_id: UUID) -> None:
pass
class ConsoleEmailNotifier(EmailNotifier):
"""Email adapter that just prints — for development."""
def send_order_confirmation(self, customer_email: str, order_id: UUID) -> None:
print(f"[EMAIL] Sending confirmation for order {order_id} to {customer_email}")
class SendGridEmailNotifier(EmailNotifier):
"""Real email adapter using SendGrid."""
def __init__(self, api_key: str) -> None:
self._api_key = api_key
def send_order_confirmation(self, customer_email: str, order_id: UUID) -> None:
# Actual SendGrid API call here
pass
class InMemoryEventPublisher(EventPublisher):
"""Event publisher for testing — stores events in a list."""
def __init__(self) -> None:
self.published_events: List[DomainEvent] = []
def publish(self, event: DomainEvent) -> None:
self.published_events.append(event)
print(f"[EVENT] {type(event).__name__}: {event}")
class RabbitMQEventPublisher(EventPublisher):
"""Production event publisher using RabbitMQ."""
def __init__(self, amqp_url: str) -> None:
self._amqp_url = amqp_url
def publish(self, event: DomainEvent) -> None:
# Serialize event and publish to exchange
pass
Combining DDD + Hexagonal Architecture¶
Application Services¶
Application Services are the inside of the hexagon. They orchestrate the domain to fulfill use cases. They are thin — they don't contain domain logic, they coordinate it.
class OrderApplicationService(OrderApplicationPort):
"""
Application service: implements the driving port.
Orchestrates domain objects and calls driven ports.
"""
def __init__(
self,
order_repo: OrderRepository,
customer_repo: CustomerRepository,
event_publisher: EventPublisher,
email_notifier: EmailNotifier,
) -> None:
self._orders = order_repo
self._customers = customer_repo
self._events = event_publisher
self._email = email_notifier
def create_order(self, cmd: CreateOrderCommand) -> UUID:
# Load customer (validate they exist)
customer = self._customers.find_by_id(cmd.customer_id)
if customer is None:
raise ValueError(f"Customer {cmd.customer_id} not found")
# Create domain object
address = Address(
street=cmd.shipping_address_street,
city=cmd.shipping_address_city,
country=cmd.shipping_address_country,
postal_code=cmd.shipping_address_postal,
)
order = Order(customer_id=customer.id, shipping_address=address)
# Persist
self._orders.save(order)
return order.id
def add_line_to_order(self, cmd: AddOrderLineCommand) -> None:
order = self._load_order(cmd.order_id)
order.add_line(
product_id=cmd.product_id,
product_name=cmd.product_name,
quantity=cmd.quantity,
unit_price=Money(cmd.unit_price_amount, cmd.unit_price_currency),
)
self._orders.save(order)
def confirm_order(self, order_id: UUID) -> None:
order = self._load_order(order_id)
customer = self._customers.find_by_id(order.customer_id)
order.confirm()
# Dispatch domain events
for event in order.domain_events:
self._events.publish(event)
if isinstance(event, OrderConfirmed):
self._email.send_order_confirmation(customer.email, order.id)
customer.add_loyalty_points(len(order.lines) * 10)
self._customers.save(customer)
order.clear_events()
self._orders.save(order)
def cancel_order(self, order_id: UUID) -> None:
order = self._load_order(order_id)
order.cancel()
for event in order.domain_events:
self._events.publish(event)
order.clear_events()
self._orders.save(order)
def get_order_summary(self, order_id: UUID) -> Optional[OrderSummaryDTO]:
order = self._orders.find_by_id(order_id)
if order is None:
return None
return OrderSummaryDTO(
order_id=str(order.id),
customer_id=str(order.customer_id),
status=order.status.value,
total_amount=order.total.amount,
total_currency=order.total.currency,
line_count=len(order.lines),
)
def _load_order(self, order_id: UUID) -> Order:
order = self._orders.find_by_id(order_id)
if order is None:
raise ValueError(f"Order {order_id} not found")
return order
Dependency Injection / Composition Root¶
Wire everything together in one place — the composition root. This is typically the application's entry point.
# composition_root.py
def build_application(settings: dict) -> OrderApplicationPort:
"""
The composition root: wire all adapters to their ports.
Swap adapters here for different environments.
"""
if settings.get("env") == "production":
order_repo = PostgresOrderRepository(settings["db_url"])
email_notifier = SendGridEmailNotifier(settings["sendgrid_key"])
event_publisher = RabbitMQEventPublisher(settings["amqp_url"])
else:
order_repo = InMemoryOrderRepository()
email_notifier = ConsoleEmailNotifier()
event_publisher = InMemoryEventPublisher()
customer_repo = InMemoryCustomerRepository()
return OrderApplicationService(
order_repo=order_repo,
customer_repo=customer_repo,
event_publisher=event_publisher,
email_notifier=email_notifier,
)
# main.py
if __name__ == "__main__":
import os
settings = {
"env": os.getenv("APP_ENV", "development"),
"db_url": os.getenv("DATABASE_URL", ""),
"sendgrid_key": os.getenv("SENDGRID_API_KEY", ""),
"amqp_url": os.getenv("AMQP_URL", ""),
}
service = build_application(settings)
# Wire into HTTP framework
http_adapter = FastAPIOrderAdapter(service)
Project Structure¶
A recommended directory layout that reflects the architecture:
order_service/
│
├── domain/ # Pure Python, zero dependencies
│ ├── __init__.py
│ ├── model/
│ │ ├── __init__.py
│ │ ├── order.py # Order aggregate, OrderLine, OrderStatus
│ │ ├── customer.py # Customer entity
│ │ └── value_objects.py # Money, Address
│ ├── events/
│ │ ├── __init__.py
│ │ └── order_events.py # OrderConfirmed, OrderCancelled, etc.
│ ├── services/
│ │ ├── __init__.py
│ │ └── pricing_service.py # Domain services
│ └── repositories/
│ ├── __init__.py
│ ├── order_repository.py # Abstract OrderRepository
│ └── customer_repository.py # Abstract CustomerRepository
│
├── application/ # Use cases, orchestration
│ ├── __init__.py
│ ├── ports/
│ │ ├── __init__.py
│ │ ├── incoming.py # Driving ports (commands, DTOs, port interfaces)
│ │ └── outgoing.py # Driven ports (EventPublisher, EmailNotifier)
│ └── services/
│ ├── __init__.py
│ └── order_service.py # OrderApplicationService
│
├── adapters/ # All infrastructure code
│ ├── __init__.py
│ ├── primary/ # Driving adapters
│ │ ├── __init__.py
│ │ ├── http/
│ │ │ ├── __init__.py
│ │ │ ├── routes.py # FastAPI/Flask routes
│ │ │ └── schemas.py # Pydantic request/response models
│ │ └── cli/
│ │ └── commands.py # CLI commands (Click/Typer)
│ └── secondary/ # Driven adapters
│ ├── __init__.py
│ ├── persistence/
│ │ ├── __init__.py
│ │ ├── in_memory.py # InMemoryOrderRepository
│ │ ├── postgres.py # PostgresOrderRepository
│ │ └── models.py # SQLAlchemy ORM models (separate from domain)
│ ├── messaging/
│ │ ├── __init__.py
│ │ └── rabbitmq.py # RabbitMQEventPublisher
│ └── notifications/
│ ├── __init__.py
│ ├── console.py # ConsoleEmailNotifier
│ └── sendgrid.py # SendGridEmailNotifier
│
├── composition_root.py # Wire everything together
├── main.py # Entry point
└── tests/
├── unit/
│ ├── test_order.py # Test domain model in isolation
│ └── test_pricing_service.py
├── integration/
│ └── test_order_service.py # Test application service with in-memory adapters
└── e2e/
└── test_http_api.py # Test full HTTP stack
Key constraint: dependencies only flow inward. adapters/ imports from application/ and domain/. application/ imports from domain/. domain/ imports from nothing (except stdlib).
Full Working Example¶
Here's a complete runnable example tying it all together:
# run_example.py
from uuid import uuid4
# Setup
order_repo = InMemoryOrderRepository()
customer_repo = InMemoryCustomerRepository()
event_publisher = InMemoryEventPublisher()
email_notifier = ConsoleEmailNotifier()
service = OrderApplicationService(
order_repo=order_repo,
customer_repo=customer_repo,
event_publisher=event_publisher,
email_notifier=email_notifier,
)
# Seed a customer
customer = Customer(
name="Alice Smith",
email="alice@example.com",
address=Address("123 Main St", "Springfield", "US", "12345"),
)
customer_repo.save(customer)
# Create an order
order_id = service.create_order(CreateOrderCommand(
customer_id=customer.id,
shipping_address_street="456 Elm St",
shipping_address_city="Shelbyville",
shipping_address_country="US",
shipping_address_postal="67890",
))
print(f"Created order: {order_id}")
# Add line items
product_id = uuid4()
service.add_line_to_order(AddOrderLineCommand(
order_id=order_id,
product_id=product_id,
product_name="Python Cookbook",
quantity=2,
unit_price_amount=29.99,
unit_price_currency="USD",
))
# Check summary
summary = service.get_order_summary(order_id)
print(f"Order status: {summary.status}, total: {summary.total_amount} {summary.total_currency}")
# Order status: pending, total: 59.98 USD
# Confirm it
service.confirm_order(order_id)
# [EMAIL] Sending confirmation for order ... to alice@example.com
# [EVENT] OrderConfirmed: ...
summary = service.get_order_summary(order_id)
print(f"Order status after confirm: {summary.status}")
# Order status after confirm: confirmed
# Try to add a line to a confirmed order (should raise)
try:
service.add_line_to_order(AddOrderLineCommand(
order_id=order_id,
product_id=uuid4(),
product_name="Extra Book",
quantity=1,
unit_price_amount=15.00,
unit_price_currency="USD",
))
except ValueError as e:
print(f"Expected error: {e}")
# Expected error: Cannot modify order in status confirmed
Testing Strategy¶
The architecture makes testing elegant. Each layer is tested independently.
Unit Tests — Domain Layer (No Mocks Needed)¶
import pytest
from uuid import uuid4
def make_address() -> Address:
return Address("1 St", "City", "US", "11111")
class TestOrder:
def test_cannot_confirm_empty_order(self):
order = Order(customer_id=uuid4(), shipping_address=make_address())
with pytest.raises(ValueError, match="empty order"):
order.confirm()
def test_add_line_increases_total(self):
order = Order(customer_id=uuid4(), shipping_address=make_address())
order.add_line(uuid4(), "Widget", 3, Money(10.00, "USD"))
assert order.total == Money(30.00, "USD")
def test_confirm_emits_event(self):
order = Order(customer_id=uuid4(), shipping_address=make_address())
order.add_line(uuid4(), "Widget", 1, Money(5.00, "USD"))
order.confirm()
assert len(order.domain_events) == 1
assert isinstance(order.domain_events[0], OrderConfirmed)
def test_cannot_modify_confirmed_order(self):
order = Order(customer_id=uuid4(), shipping_address=make_address())
order.add_line(uuid4(), "Widget", 1, Money(5.00, "USD"))
order.confirm()
with pytest.raises(ValueError, match="Cannot modify"):
order.add_line(uuid4(), "Another", 1, Money(5.00, "USD"))
def test_cancel_shipped_order_raises(self):
order = Order(customer_id=uuid4(), shipping_address=make_address())
order.add_line(uuid4(), "Widget", 1, Money(5.00, "USD"))
order.confirm()
order.status = OrderStatus.SHIPPED
with pytest.raises(ValueError, match="Cannot cancel"):
order.cancel()
class TestMoney:
def test_add_same_currency(self):
assert Money(1.00, "USD").add(Money(2.50, "USD")) == Money(3.50, "USD")
def test_add_different_currency_raises(self):
with pytest.raises(ValueError):
Money(1.00, "USD").add(Money(1.00, "EUR"))
def test_negative_amount_raises(self):
with pytest.raises(ValueError):
Money(-1.00, "USD")
Integration Tests — Application Service (In-Memory Adapters)¶
class TestOrderApplicationService:
"""Test use cases using in-memory adapters — no I/O needed."""
def setup_method(self):
self.order_repo = InMemoryOrderRepository()
self.customer_repo = InMemoryCustomerRepository()
self.events = InMemoryEventPublisher()
self.email = ConsoleEmailNotifier()
self.service = OrderApplicationService(
self.order_repo, self.customer_repo, self.events, self.email
)
# Seed a customer
self.customer = Customer(
"Bob", "bob@test.com", make_address()
)
self.customer_repo.save(self.customer)
def test_full_order_lifecycle(self):
# Create
order_id = self.service.create_order(CreateOrderCommand(
customer_id=self.customer.id,
shipping_address_street="1 Test St",
shipping_address_city="Testville",
shipping_address_country="US",
shipping_address_postal="00001",
))
assert order_id is not None
# Add line
self.service.add_line_to_order(AddOrderLineCommand(
order_id=order_id,
product_id=uuid4(),
product_name="Gadget",
quantity=2,
unit_price_amount=25.00,
unit_price_currency="USD",
))
# Confirm
self.service.confirm_order(order_id)
# Verify state
summary = self.service.get_order_summary(order_id)
assert summary.status == "confirmed"
assert summary.total_amount == 50.00
# Verify events were published
assert len(self.events.published_events) == 1
assert isinstance(self.events.published_events[0], OrderConfirmed)
def test_cancel_order(self):
order_id = self.service.create_order(CreateOrderCommand(
customer_id=self.customer.id,
shipping_address_street="1 St",
shipping_address_city="C",
shipping_address_country="US",
shipping_address_postal="00001",
))
self.service.add_line_to_order(AddOrderLineCommand(
order_id=order_id, product_id=uuid4(),
product_name="Item", quantity=1,
unit_price_amount=10.00, unit_price_currency="USD",
))
self.service.cancel_order(order_id)
summary = self.service.get_order_summary(order_id)
assert summary.status == "cancelled"
def test_unknown_customer_raises(self):
with pytest.raises(ValueError, match="not found"):
self.service.create_order(CreateOrderCommand(
customer_id=uuid4(), # Non-existent
shipping_address_street="1 St",
shipping_address_city="C",
shipping_address_country="US",
shipping_address_postal="00001",
))
Summary & Best Practices¶
DDD Quick Reference¶
| Concept | When to Use | Key Rule |
|---|---|---|
| Value Object | Descriptive attributes with no identity | Always immutable |
| Entity | Things with a lifecycle and identity | Equality by ID |
| Aggregate | Consistency boundary around related objects | Modify only through root |
| Domain Event | Record that something happened | Immutable, past-tense name |
| Repository | Access aggregates from storage | Interface in domain, implementation in infra |
| Domain Service | Logic spanning multiple aggregates | Stateless |
| Bounded Context | Large domain with distinct sub-models | Own schema, own team, communicate via contracts |
Hexagonal Architecture Quick Reference¶
| Concept | Role | Lives In |
|---|---|---|
| Driving Port | What use cases the app exposes | application/ports/incoming.py |
| Driven Port | What infrastructure the app needs | application/ports/outgoing.py |
| Primary Adapter | HTTP, CLI, test harness | adapters/primary/ |
| Secondary Adapter | DB, email, queue | adapters/secondary/ |
| Application Service | Orchestrates domain to fulfill use cases | application/services/ |
| Composition Root | Wires ports to adapters | composition_root.py |
The Golden Rules¶
- Domain first — write domain objects before thinking about databases or HTTP
- Dependency inversion — domain and application define interfaces; adapters implement them
- One direction — dependencies point inward:
adapters → application → domain - No leakage — domain objects never import from adapters or the framework
- Test at the right level — unit test domain logic; integration test use cases with in-memory adapters; e2e test the full stack sparingly
- Aggregates are consistency units — if two things must change together, they belong in the same aggregate
- Repositories per aggregate root only — never a repository for a non-root entity
- Ubiquitous language — use business terms in class names, method names, and variables, not technical jargon
Common Pitfalls to Avoid¶
- Anemic domain model — entities with only getters/setters, all logic in services. Put behaviour on your domain objects.
- Fat application services — if your service has domain logic (
if price > 100: apply_discount()), move it to the domain. - Repository returning raw SQL rows — repositories must return fully-constructed domain aggregates.
- Domain events with mutable state — events describe the past; make them
frozen=True. - Putting ORM models in the domain — SQLAlchemy models belong in the infrastructure layer, not the domain.
This primer covers the core patterns. For deeper study, refer to Eric Evans' "Domain-Driven Design" (2003), Vaughn Vernon's "Implementing Domain-Driven Design" (2013), and Alistair Cockburn's original Hexagonal Architecture article at alistair.cockburn.us.