Domain-Driven Design: Building Domain-Centric Applications

# Sales Context
class Customer:
    """Customer in sales context - focused on purchasing."""
    def __init__(self, customer_id: str, name: str, credit_limit: Decimal):
        self.id = customer_id
        self.name = name
        self.credit_limit = credit_limit
        self.orders: List[Order] = []
    
    def can_place_order(self, order_total: Decimal) -> bool:
        """Check if customer can place order based on credit."""
        current_debt = sum(o.total for o in self.orders if not o.is_paid)
        return current_debt + order_total <= self.credit_limit

# Support Context
class Customer:
    """Customer in support context - focused on service."""
    def __init__(self, customer_id: str, name: str, support_tier: str):
        self.id = customer_id
        self.name = name
        self.support_tier = support_tier  # basic, premium, enterprise
        self.tickets: List[SupportTicket] = []
    
    def get_response_sla(self) -> timedelta:
        """Get SLA based on support tier."""
        sla_map = {
            'basic': timedelta(hours=48),
            'premium': timedelta(hours=24),
            'enterprise': timedelta(hours=4)
        }
        return sla_map[self.support_tier]

# Shipping Context
class Customer:
    """Customer in shipping context - focused on delivery."""
    def __init__(self, customer_id: str, name: str, addresses: List[Address]):
        self.id = customer_id
        self.name = name
        self.addresses = addresses
        self.preferred_carrier: Optional[str] = None
    
    def get_default_shipping_address(self) -> Address:
        """Get default address for shipping."""
        return next((a for a in self.addresses if a.is_default), self.addresses[0])

# Anti-Corruption Layer (ACL)
# Protects domain model from external system changes

class LegacyCustomerDTO:
    """Data from legacy system."""
    cust_no: str
    full_name: str
    credit_amt: str  # String in legacy system!
    status_cd: str   # Cryptic codes

class CustomerAdapter:
    """ACL: Translate legacy data to domain model."""
    
    def to_domain(self, dto: LegacyCustomerDTO) -> Customer:
        """Convert legacy DTO to domain Customer."""
        return Customer(
            customer_id=dto.cust_no,
            name=dto.full_name,
            credit_limit=Decimal(dto.credit_amt),
            status=self._translate_status(dto.status_cd)
        )
    
    def _translate_status(self, status_code: str) -> CustomerStatus:
        """Translate cryptic codes to domain enum."""
        mapping = {
            'A': CustomerStatus.ACTIVE,
            'S': CustomerStatus.SUSPENDED,
            'C': CustomerStatus.CLOSED
        }
        return mapping.get(status_code, CustomerStatus.UNKNOWN)
    
    def from_domain(self, customer: Customer) -> LegacyCustomerDTO:
        """Convert domain Customer to legacy DTO."""
        return LegacyCustomerDTO(
            cust_no=customer.id,
            full_name=customer.name,
            credit_amt=str(customer.credit_limit),
            status_cd=self._translate_status_to_code(customer.status)
        )

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import uuid

class Entity:
    """Base entity with identity and equality based on ID."""
    
    def __init__(self, id: str = None):
        self.id = id or str(uuid.uuid4())
    
    def __eq__(self, other):
        if not isinstance(other, type(self)):
            return False
        return self.id == other.id
    
    def __hash__(self):
        return hash(self.id)

class User(Entity):
    """User entity - identity matters more than attributes."""
    
    def __init__(self, email: str, name: str, id: str = None):
        super().__init__(id)
        self._email = email
        self._name = name
        self._created_at = datetime.utcnow()
        self._is_active = True
        self._email_verified = False
    
    @property
    def email(self) -> str:
        return self._email
    
    @property
    def name(self) -> str:
        return self._name
    
    @property
    def is_active(self) -> bool:
        return self._is_active
    
    def change_email(self, new_email: str):
        """Change email - requires re-verification."""
        if "@" not in new_email:
            raise ValueError("Invalid email format")
        
        self._email = new_email
        self._email_verified = False
    
    def verify_email(self):
        """Mark email as verified."""
        self._email_verified = True
    
    def activate(self):
        """Activate user account."""
        if not self._email_verified:
            raise ValueError("Cannot activate unverified account")
        self._is_active = True
    
    def deactivate(self):
        """Deactivate user account."""
        self._is_active = False

# Two users with same attributes but different IDs are different
user1 = User(email="[email protected]", name="John", id="1")
user2 = User(email="[email protected]", name="John", id="2")
assert user1 != user2  # Different entities

# Same user after attribute change is still the same entity
user1.change_email("[email protected]")
user1_reloaded = User(email="[email protected]", name="John", id="1")
assert user1 == user1_reloaded  # Same entity

@dataclass(frozen=True)
class Money:
    """Value object - immutable and defined by values."""
    amount: Decimal
    currency: str
    
    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Amount cannot be negative")
        if not self.currency or len(self.currency) != 3:
            raise ValueError("Currency must be 3-letter code")
    
    def add(self, other: 'Money') -> 'Money':
        """Add money - returns new instance."""
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return Money(self.amount + other.amount, self.currency)
    
    def multiply(self, factor: Decimal) -> 'Money':
        """Multiply money - returns new instance."""
        return Money(self.amount * factor, self.currency)
    
    def __str__(self):
        return f"{self.amount} {self.currency}"

@dataclass(frozen=True)
class Address:
    """Value object for address."""
    street: str
    city: str
    state: str
    zip_code: str
    country: str
    
    def __post_init__(self):
        if not self.street or not self.city:
            raise ValueError("Street and city are required")
    
    def is_in_country(self, country: str) -> bool:
        return self.country.upper() == country.upper()

@dataclass(frozen=True)
class DateRange:
    """Value object for date range."""
    start: datetime
    end: datetime
    
    def __post_init__(self):
        if self.start >= self.end:
            raise ValueError("Start must be before end")
    
    def contains(self, date: datetime) -> bool:
        """Check if date is within range."""
        return self.start <= date <= self.end
    
    def overlaps(self, other: 'DateRange') -> bool:
        """Check if ranges overlap."""
        return self.start <= other.end and other.start <= self.end
    
    def duration(self) -> timedelta:
        """Get duration of range."""
        return self.end - self.start

# Value objects are interchangeable
price1 = Money(Decimal('99.99'), 'USD')
price2 = Money(Decimal('99.99'), 'USD')
assert price1 == price2  # Same value = equal

# Immutability
total = price1.add(price2)  # Returns new instance
assert price1.amount == Decimal('99.99')  # Original unchanged

from typing import List
from decimal import Decimal

class OrderLine:
    """Entity within Order aggregate."""
    
    def __init__(self, product_id: str, product_name: str, 
                 quantity: int, unit_price: Money):
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        
        self.product_id = product_id
        self.product_name = product_name
        self.quantity = quantity
        self.unit_price = unit_price
    
    @property
    def total(self) -> Money:
        """Calculate line total."""
        return self.unit_price.multiply(Decimal(self.quantity))
    
    def change_quantity(self, new_quantity: int):
        """Change quantity - must be positive."""
        if new_quantity <= 0:
            raise ValueError("Quantity must be positive")
        self.quantity = new_quantity

class Order(Entity):
    """Aggregate root - controls access to OrderLines."""
    
    def __init__(self, customer_id: str, id: str = None):
        super().__init__(id)
        self._customer_id = customer_id
        self._lines: List[OrderLine] = []
        self._status = OrderStatus.DRAFT
        self._created_at = datetime.utcnow()
        self._submitted_at: Optional[datetime] = None
    
    @property
    def customer_id(self) -> str:
        return self._customer_id
    
    @property
    def status(self) -> 'OrderStatus':
        return self._status
    
    @property
    def lines(self) -> List[OrderLine]:
        """Return copy to prevent external modification."""
        return self._lines.copy()
    
    def add_line(self, product_id: str, product_name: str,
                 quantity: int, unit_price: Money):
        """Add line to order - only through aggregate root."""
        if self._status != OrderStatus.DRAFT:
            raise ValueError("Cannot modify submitted order")
        
        # Check if product already in order
        existing = next((l for l in self._lines if l.product_id == product_id), None)
        
        if existing:
            existing.change_quantity(existing.quantity + quantity)
        else:
            line = OrderLine(product_id, product_name, quantity, unit_price)
            self._lines.append(line)
    
    def remove_line(self, product_id: str):
        """Remove line from order."""
        if self._status != OrderStatus.DRAFT:
            raise ValueError("Cannot modify submitted order")
        
        self._lines = [l for l in self._lines if l.product_id != product_id]
    
    def change_line_quantity(self, product_id: str, new_quantity: int):
        """Change quantity of existing line."""
        if self._status != OrderStatus.DRAFT:
            raise ValueError("Cannot modify submitted order")
        
        line = next((l for l in self._lines if l.product_id == product_id), None)
        if not line:
            raise ValueError(f"Product {product_id} not in order")
        
        line.change_quantity(new_quantity)
    
    def submit(self):
        """Submit order - transition to submitted state."""
        if self._status != OrderStatus.DRAFT:
            raise ValueError("Order already submitted")
        
        if not self._lines:
            raise ValueError("Cannot submit empty order")
        
        self._status = OrderStatus.SUBMITTED
        self._submitted_at = datetime.utcnow()
    
    def cancel(self):
        """Cancel order."""
        if self._status == OrderStatus.SHIPPED:
            raise ValueError("Cannot cancel shipped order")
        
        if self._status == OrderStatus.CANCELLED:
            raise ValueError("Order already cancelled")
        
        self._status = OrderStatus.CANCELLED
    
    def ship(self):
        """Mark order as shipped."""
        if self._status != OrderStatus.SUBMITTED:
            raise ValueError("Can only ship submitted orders")
        
        self._status = OrderStatus.SHIPPED
    
    @property
    def total(self) -> Money:
        """Calculate order total."""
        if not self._lines:
            return Money(Decimal('0'), 'USD')
        
        # All lines must have same currency
        currency = self._lines[0].unit_price.currency
        total_amount = sum(line.total.amount for line in self._lines)
        return Money(total_amount, currency)

class OrderStatus:
    """Order status enum."""
    DRAFT = 'draft'
    SUBMITTED = 'submitted'
    SHIPPED = 'shipped'
    CANCELLED = 'cancelled'

# Usage - all modifications go through aggregate root
order = Order(customer_id='cust-123')
order.add_line('prod-1', 'Widget', 2, Money(Decimal('10.00'), 'USD'))
order.add_line('prod-2', 'Gadget', 1, Money(Decimal('25.00'), 'USD'))
order.change_line_quantity('prod-1', 3)
order.submit()

# Cannot modify lines directly - they're encapsulated
# order.lines[0].quantity = 10  # Won't work - lines() returns copy

from dataclasses import dataclass
from datetime import datetime

@dataclass
class DomainEvent:
    """Domain event - something that happened."""
    event_id: str
    event_type: str
    occurred_at: datetime
    aggregate_id: str

@dataclass
class OrderSubmitted(DomainEvent):
    order_id: str
    user_id: str
    total: float

class Order(AggregateRoot):
    """Aggregate that publishes events."""
    
    def __init__(self, user_id: str):
        super().__init__()
        self._user_id = user_id
        self._items = []
        self._status = "pending"
        self._events = []
    
    def submit(self):
        self._status = "submitted"
        
        event = OrderSubmitted(
            event_id=str(uuid.uuid4()),
            event_type="order.submitted",
            occurred_at=datetime.utcnow(),
            aggregate_id=self.id,
            order_id=self.id,
            user_id=self._user_id,
            total=self.total
        )
        self._events.append(event)
    
    def pull_events(self) -> List[DomainEvent]:
        events = self._events.copy()
        self._events.clear()
        return events

from abc import ABC, abstractmethod
from typing import Optional, List

class OrderRepository(ABC):
    """Repository interface - domain layer."""
    
    @abstractmethod
    def get(self, order_id: str) -> Optional[Order]:
        """Get order by ID."""
        pass
    
    @abstractmethod
    def save(self, order: Order) -> None:
        """Save order."""
        pass
    
    @abstractmethod
    def find_by_customer(self, customer_id: str) -> List[Order]:
        """Find orders by customer."""
        pass
    
    @abstractmethod
    def find_pending_orders(self) -> List[Order]:
        """Find all pending orders."""
        pass

class SQLOrderRepository(OrderRepository):
    """Repository implementation - infrastructure layer."""
    
    def __init__(self, session):
        self.session = session
    
    def get(self, order_id: str) -> Optional[Order]:
        """Load order from database."""
        row = self.session.query(OrderModel).filter_by(id=order_id).first()
        
        if not row:
            return None
        
        # Reconstruct domain object from database model
        order = Order(customer_id=row.customer_id, id=row.id)
        order._status = row.status
        order._created_at = row.created_at
        order._submitted_at = row.submitted_at
        
        # Load order lines
        for line_row in row.lines:
            order._lines.append(OrderLine(
                product_id=line_row.product_id,
                product_name=line_row.product_name,
                quantity=line_row.quantity,
                unit_price=Money(Decimal(line_row.unit_price), line_row.currency)
            ))
        
        return order
    
    def save(self, order: Order) -> None:
        """Save order to database."""
        # Find or create database model
        order_model = self.session.query(OrderModel).filter_by(id=order.id).first()
        
        if not order_model:
            order_model = OrderModel(id=order.id)
            self.session.add(order_model)
        
        # Update attributes
        order_model.customer_id = order.customer_id
        order_model.status = order.status
        order_model.created_at = order._created_at
        order_model.submitted_at = order._submitted_at
        
        # Update lines
        order_model.lines.clear()
        for line in order.lines:
            line_model = OrderLineModel(
                product_id=line.product_id,
                product_name=line.product_name,
                quantity=line.quantity,
                unit_price=str(line.unit_price.amount),
                currency=line.unit_price.currency
            )
            order_model.lines.append(line_model)
        
        self.session.commit()
    
    def find_by_customer(self, customer_id: str) -> List[Order]:
        """Find orders by customer."""
        rows = self.session.query(OrderModel).filter_by(customer_id=customer_id).all()
        return [self.get(row.id) for row in rows]
    
    def find_pending_orders(self) -> List[Order]:
        """Find pending orders."""
        rows = self.session.query(OrderModel).filter_by(status=OrderStatus.SUBMITTED).all()
        return [self.get(row.id) for row in rows]

# Usage - domain layer doesn't know about database
class OrderService:
    def __init__(self, order_repo: OrderRepository):
        self.order_repo = order_repo
    
    def submit_order(self, order_id: str):
        order = self.order_repo.get(order_id)
        if not order:
            raise ValueError("Order not found")
        
        order.submit()
        self.order_repo.save(order)

from dataclasses import dataclass
from datetime import datetime
from typing import List
import uuid

@dataclass
class DomainEvent:
    """Base domain event."""
    event_id: str
    event_type: str
    occurred_at: datetime
    aggregate_id: str
    aggregate_type: str

@dataclass
class OrderSubmitted(DomainEvent):
    """Event: Order was submitted."""
    customer_id: str
    order_total: Money
    line_count: int

@dataclass
class OrderShipped(DomainEvent):
    """Event: Order was shipped."""
    customer_id: str
    tracking_number: str
    carrier: str

class Order(Entity):
    """Aggregate that publishes domain events."""
    
    def __init__(self, customer_id: str, id: str = None):
        super().__init__(id)
        self._customer_id = customer_id
        self._lines: List[OrderLine] = []
        self._status = OrderStatus.DRAFT
        self._domain_events: List[DomainEvent] = []
    
    def submit(self):
        """Submit order and raise event."""
        if self._status != OrderStatus.DRAFT:
            raise ValueError("Order already submitted")
        
        if not self._lines:
            raise ValueError("Cannot submit empty order")
        
        self._status = OrderStatus.SUBMITTED
        self._submitted_at = datetime.utcnow()
        
        # Raise domain event
        event = OrderSubmitted(
            event_id=str(uuid.uuid4()),
            event_type='OrderSubmitted',
            occurred_at=datetime.utcnow(),
            aggregate_id=self.id,
            aggregate_type='Order',
            customer_id=self._customer_id,
            order_total=self.total,
            line_count=len(self._lines)
        )
        self._domain_events.append(event)
    
    def ship(self, tracking_number: str, carrier: str):
        """Ship order and raise event."""
        if self._status != OrderStatus.SUBMITTED:
            raise ValueError("Can only ship submitted orders")
        
        self._status = OrderStatus.SHIPPED
        
        event = OrderShipped(
            event_id=str(uuid.uuid4()),
            event_type='OrderShipped',
            occurred_at=datetime.utcnow(),
            aggregate_id=self.id,
            aggregate_type='Order',
            customer_id=self._customer_id,
            tracking_number=tracking_number,
            carrier=carrier
        )
        self._domain_events.append(event)
    
    def pull_domain_events(self) -> List[DomainEvent]:
        """Get and clear domain events."""
        events = self._domain_events.copy()
        self._domain_events.clear()
        return events

# Event handlers
class OrderEventHandler:
    """Handle order events."""
    
    def __init__(self, email_service, inventory_service):
        self.email_service = email_service
        self.inventory_service = inventory_service
    
    def handle(self, event: DomainEvent):
        """Dispatch event to appropriate handler."""
        if isinstance(event, OrderSubmitted):
            self._handle_order_submitted(event)
        elif isinstance(event, OrderShipped):
            self._handle_order_shipped(event)
    
    def _handle_order_submitted(self, event: OrderSubmitted):
        """Handle order submitted event."""
        # Send confirmation email
        self.email_service.send_order_confirmation(
            customer_id=event.customer_id,
            order_id=event.aggregate_id,
            total=event.order_total
        )
        
        # Reserve inventory
        self.inventory_service.reserve_for_order(event.aggregate_id)
    
    def _handle_order_shipped(self, event: OrderShipped):
        """Handle order shipped event."""
        # Send shipping notification
        self.email_service.send_shipping_notification(
            customer_id=event.customer_id,
            tracking_number=event.tracking_number,
            carrier=event.carrier
        )

# Usage in application service
class OrderApplicationService:
    def __init__(self, order_repo: OrderRepository, event_handler: OrderEventHandler):
        self.order_repo = order_repo
        self.event_handler = event_handler
    
    def submit_order(self, order_id: str):
        """Submit order and publish events."""
        order = self.order_repo.get(order_id)
        if not order:
            raise ValueError("Order not found")
        
        order.submit()
        self.order_repo.save(order)
        
        # Publish domain events
        for event in order.pull_domain_events():
            self.event_handler.handle(event)

class PricingService:
    """Domain service for pricing calculations."""
    
    def __init__(self, discount_repo, tax_calculator):
        self.discount_repo = discount_repo
        self.tax_calculator = tax_calculator
    
    def calculate_order_total(self, order: Order, customer: Customer) -> Money:
        """Calculate order total with discounts and tax."""
        # Subtotal
        subtotal = order.total
        
        # Apply customer discounts
        discount = self.discount_repo.get_customer_discount(customer.id)
        if discount:
            discount_amount = subtotal.multiply(discount.percentage)
            subtotal = subtotal.add(discount_amount.multiply(Decimal('-1')))
        
        # Calculate tax
        tax_amount = self.tax_calculator.calculate_tax(
            amount=subtotal,
            customer_address=customer.billing_address
        )
        
        # Total
        return subtotal.add(tax_amount)

class TransferService:
    """Domain service for money transfers between accounts."""
    
    def transfer(self, from_account: BankAccount, to_account: BankAccount,
                 amount: Money):
        """Transfer money between accounts."""
        if from_account.currency != to_account.currency:
            raise ValueError("Cannot transfer between different currencies")
        
        if from_account.balance < amount:
            raise ValueError("Insufficient funds")
        
        # Withdraw from source
        from_account.withdraw(amount, f"Transfer to {to_account.id}")
        
        # Deposit to destination
        to_account.deposit(amount, f"Transfer from {from_account.id}")

class UserRecord:
    def persist_to_db(self):
        pass
    
    def fetch_related_records(self):
        pass

class Customer:
    def save(self):
        pass
    
    def get_orders(self):
        pass

# Domain Layer - Pure business logic
class Order(Entity):
    def submit(self):
        if not self._lines:
            raise ValueError("Cannot submit empty order")
        self._status = OrderStatus.SUBMITTED

# Application Layer - Use case orchestration
class OrderApplicationService:
    def __init__(self, order_repo, email_service, inventory_service):
        self.order_repo = order_repo
        self.email_service = email_service
        self.inventory_service = inventory_service
    
    def submit_order(self, order_id: str):
        """Submit order use case."""
        # Load aggregate
        order = self.order_repo.get(order_id)
        
        # Execute domain logic
        order.submit()
        
        # Save changes
        self.order_repo.save(order)
        
        # Side effects
        self.email_service.send_confirmation(order.customer_id, order.id)
        self.inventory_service.reserve(order.id)

# Infrastructure Layer - Technical implementation
class SQLOrderRepository(OrderRepository):
    def save(self, order: Order):
        # Database-specific code
        pass

# Presentation Layer - HTTP API
@app.post("/orders/{order_id}/submit")
def submit_order_endpoint(order_id: str, service: OrderApplicationService):
    try:
        service.submit_order(order_id)
        return {"status": "success"}
    except ValueError as e:
        return {"status": "error", "message": str(e)}, 400

# Bad - Anemic
class Order:
    def get_status(self):
        return self.status
    
    def set_status(self, status):
        self.status = status

class OrderService:
    def submit_order(self, order):
        if not order.get_items():
            raise ValueError("Empty order")
        order.set_status("submitted")

# Good - Rich domain model
class Order:
    def submit(self):
        if not self._items:
            raise ValueError("Empty order")
        self._status = OrderStatus.SUBMITTED

# Bad - Logic in application layer
class OrderService:
    def submit_order(self, order_id):
        order = self.repo.get(order_id)
        if not order.items:  # Business rule in service!
            raise ValueError("Empty order")
        order.status = "submitted"

# Good - Logic in domain
class Order:
    def submit(self):
        if not self._items:  # Business rule in aggregate
            raise ValueError("Empty order")
        self._status = OrderStatus.SUBMITTED

# Bad - Large aggregate
class Customer(AggregateRoot):
    orders: List[Order]  # Hundreds of orders!
    addresses: List[Address]
    payment_methods: List[PaymentMethod]
    preferences: CustomerPreferences

# Good - Separate aggregates
class Customer(AggregateRoot):
    # Only customer data
    pass

class Order(AggregateRoot):
    customer_id: str  # Reference, not embedded
    # Order data