Domain Driven Design Impl

This reference describes practical implementation patterns for domain-driven design (DDD) that apply across languages, frameworks and platforms. It covers how to structure code, define boundaries, model relationships and handle persistence in ways that reflect business concepts rather than technical constraints.

Define Bounded Contexts

Bounded contexts are distinct areas of a system with their own models and language. They typically have clear boundaries around business capabilities or organizational units.

Identify Context Boundaries

A bounded context should contain:

• A cohesive set of domain concepts
• Clear ownership of data and behavior
• Defined interfaces to other contexts
• Single responsibility for a business capability

# Example: User Management Context
src/
  user_management/
    models.py      # User, Role, Permission
    services.py    # UserService, RoleService
    repositories.py # UserRepository, RoleRepository
    api.py         # REST endpoints for user operations

Context Mapping

Contexts may relate through these patterns:

• Customer-Supplier: Supplier provides data to customer
• Partnership: Both contexts own shared concepts
• Shared Kernel: Common model between contexts
• Anti-Corruption Layer: Translates between incompatible models

Implement Domain Models

Domain models represent business concepts and their behavior. They should encapsulate business logic rather than being simple data containers.

Value Objects

Value objects are immutable and compared by value, not identity:

class Money:
    def __init__(self, amount, currency):
        self.amount = amount
        self.currency = currency
    
    def __eq__(self, other):
        return (self.amount == other.amount and 
                self.currency == other.currency)

Entities

Entities have identity that persists across time:

class User:
    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email
        self.created_at = datetime.utcnow()
    
    def update_email(self, new_email):
        # Business rule: email must be unique
        self.email = new_email

Aggregates

Aggregates are clusters of entities and value objects with a root entity:

class Order:
    def __init__(self, order_id):
        self.order_id = order_id
        self.items = []
        self.status = "pending"
    
    def add_item(self, item):
        # Aggregate root ensures business rules
        if len(self.items) >= 100:
            raise ValueError("Order cannot have more than 100 items")
        self.items.append(item)

Handle Persistence

Domain models should be decoupled from persistence mechanisms through repositories.

Repository Interface

class UserRepository:
    def save(self, user):
        pass
    
    def find_by_id(self, user_id):
        pass
    
    def find_by_email(self, email):
        pass

Repository Implementation

class SqlUserRepository:
    def __init__(self, connection):
        self.connection = connection
    
    def save(self, user):
        # SQL insert or update
        with self.connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO users (id, name, email) VALUES (%s, %s, %s)",
                (user.user_id, user.name, user.email)
            )
    
    def find_by_id(self, user_id):
        with self.connection.cursor() as cursor:
            cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            row = cursor.fetchone()
            return User(row['id'], row['name'], row['email']) if row else None

Manage Domain Events

Domain events capture significant business occurrences that other parts of the system may need to react to.

Event Definition

class OrderPlacedEvent:
    def __init__(self, order_id, customer_id, items):
        self.order_id = order_id
        self.customer_id = customer_id
        self.items = items
        self.timestamp = datetime.utcnow()

Event Publishing

class EventBus:
    def __init__(self):
        self.listeners = []
    
    def subscribe(self, listener):
        self.listeners.append(listener)
    
    def publish(self, event):
        for listener in self.listeners:
            listener(event)

Implement Application Services

Application services coordinate between domain objects and infrastructure concerns.

class OrderService:
    def __init__(self, order_repository, user_repository, event_bus):
        self.order_repository = order_repository
        self.user_repository = user_repository
        self.event_bus = event_bus
    
    def create_order(self, customer_id, items):
        # Business logic
        user = self.user_repository.find_by_id(customer_id)
        if not user:
            raise ValueError("Customer not found")
        
        order = Order(str(uuid.uuid4()))
        for item in items:
            order.add_item(item)
        
        self.order_repository.save(order)
        self.event_bus.publish(OrderPlacedEvent(
            order.order_id, customer_id, items
        ))
        return order

Handle Validation

Validation should occur at domain boundaries to ensure business rules are enforced.

Domain-Level Validation

class Order:
    def add_item(self, item):
        if len(self.items) >= 100:
            raise ValueError("Order cannot have more than 100 items")
        
        # Validate item against business rules
        if item.price <= 0:
            raise ValueError("Item price must be positive")
        
        self.items.append(item)

Implement Anti-Corruption Layer

When integrating with external systems, use anti-corruption layers to translate between incompatible models.

Example: External API Integration

class ExternalUser:
    def __init__(self, user_id, full_name, email_address):
        self.user_id = user_id
        self.full_name = full_name
        self.email_address = email_address

class UserDto:
    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email

class UserAdapter:
    @staticmethod
    def to_domain(external_user):
        return User(
            external_user.user_id,
            external_user.full_name,
            external_user.email_address
        )
    
    @staticmethod
    def to_external(domain_user):
        return ExternalUser(
            domain_user.user_id,
            domain_user.name,
            domain_user.email
        )

Manage Transaction Boundaries

Aggregate roots define transaction boundaries in domain operations.

Example: Multi-Step Operation

def process_order_payment(order_id, payment_info):
    # Begin transaction
    order = order_repo.find_by_id(order_id)
    
    # Business logic
    if order.status != "pending":
        raise ValueError("Order not pending")
    
    # Apply payment
    order.status = "paid"
    order.payment_info = payment_info
    
    # Save changes
    order_repo.save(order)
    
    # Publish event
    event_bus.publish(OrderPaidEvent(order_id, payment_info))
    
    # Commit transaction

Handle Cross-Context Communication

When one context needs to interact with another, use explicit interfaces.

Context Interface

class UserServiceInterface:
    def get_user(self, user_id):
        pass
    
    def update_user(self, user):
        pass

Implementation in Context

class UserService:
    def __init__(self, user_repository):
        self.user_repository = user_repository
    
    def get_user(self, user_id):
        return self.user_repository.find_by_id(user_id)
    
    def update_user(self, user):
        return self.user_repository.save(user)

Common Implementation Pitfalls

Mixing Concerns

Don't mix business logic with infrastructure concerns in domain models:

# Bad: Domain model knows about database
class User:
    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email
    
    def save(self, connection):  # This belongs in repository
        # Database-specific code in domain object
        pass

# Good: Domain model stays pure
class User:
    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email

Incorrect Aggregate Boundaries

Aggregates should be consistent units that must stay in sync:

# Bad: Order and Customer are separate aggregates
class Order:
    def __init__(self, order_id, customer_id):
        self.order_id = order_id
        self.customer_id = customer_id
        # Customer details embedded - but customer can change independently

# Good: Customer as separate aggregate
class Order:
    def __init__(self, order_id, customer_id):
        self.order_id = order_id
        self.customer_id = customer_id
        # Only customer ID, not full customer data

Event Sourcing Issues

Domain events should be immutable and represent facts, not commands:

# Bad: Command as event
class CreateOrderCommand:
    def __init__(self, order_id, customer_id):
        self.order_id = order_id
        self.customer_id = customer_id

# Good: Fact as event
class OrderCreatedEvent:
    def __init__(self, order_id, customer_id, timestamp):
        self.order_id = order_id
        self.customer_id = customer_id
        self.timestamp = timestamp

Repository Interface Mismatches

Repositories should define clear contracts:

# Bad: Inconsistent repository interface
class UserRepository:
    def save(self, user): pass
    def find_user(self, user_id): pass  # Mixed naming convention
    def get_by_email(self, email): pass  # Different return pattern

# Good: Consistent repository interface
class UserRepository:
    def save(self, user): pass
    def find_by_id(self, user_id): pass
    def find_by_email(self, email): pass

Testing Domain Logic

Domain objects should be testable in isolation:

def test_order_add_item():
    order = Order("123")
    item = Item("item1", 10.0)
    
    order.add_item(item)
    
    assert len(order.items) == 1
    assert order.items[0] == item

def test_order_validation():
    order = Order("123")
    
    # Add 100 items
    for i in range(100):
        order.add_item(Item(f"item{i}", 10.0))
    
    # Adding 101st item should fail
    with pytest.raises(ValueError):
        order.add_item(Item("item100", 10.0))

Error Handling

Domain-specific errors should be meaningful:

class InsufficientStockError(Exception):
    def __init__(self, item_id, requested, available):
        self.item_id = item_id
        self.requested = requested
        self.available = available
        super().__init__(f"Insufficient stock for {item_id}: requested {requested}, available {available}")

class OrderProcessingError(Exception):
    pass

Performance Considerations

Avoid N+1 queries by preloading related data:

# Bad: Multiple database calls
def get_orders_with_items(customer_id):
    orders = order_repo.find_by_customer(customer_id)
    for order in orders:
        order.items = item_repo.find_by_order(order.order_id)  # N+1 query
    
    return orders

# Good: Single query with joins
def get_orders_with_items(customer_id):
    return order_repo.find_with_items_by_customer(customer_id)

Migration Patterns

When evolving domain models:

1. Keep backward compatibility in interfaces
2. Use versioned events for data evolution
3. Implement read-side projections for performance
4. Handle schema changes gracefully during deployment

Domain-driven design implementation is about creating software that reflects business reality through careful modeling, clear boundaries and disciplined separation of concerns. The patterns described here provide concrete mechanisms to achieve this without depending on specific tooling or vendor implementations.

This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.