Domain-Driven Design: Strategic Software Architecture

# E-commerce bounded contexts
contexts = {
    "orders": {
        "responsibility": "Order management",
        "core_concepts": ["Order", "OrderLine", "OrderStatus"],
        "ubiquitous_language": {
            "order": "Customer's purchase request",
            "line_item": "Product in an order"
        }
    },
    "payments": {
        "responsibility": "Payment processing",
        "core_concepts": ["Payment", "PaymentMethod", "Transaction"],
        "ubiquitous_language": {
            "payment": "Money transfer for an order",
            "authorization": "Permission to charge"
        }
    },
    "inventory": {
        "responsibility": "Stock management",
        "core_concepts": ["Product", "Stock", "Reservation"],
        "ubiquitous_language": {
            "stock": "Available quantity of product",
            "reservation": "Temporarily held inventory"
        }
    }
}

# Relationships between contexts
context_map = {
    "orders": {
        "payments": "Customer/Supplier",  # Orders consumes Payments API
        "inventory": "Conformist",       # Inventory has simple API
        "shipping": "Anticorruption Layer"  # Complex translation
    },
    "payments": {
        "orders": "Published Language",   # Shared payment events
        "billing": "Partnership"           # Close cooperation
    }
}

# Entity with identity
class Order:
    def __init__(self, order_id: OrderId, customer_id: CustomerId):
        self._id = order_id
        self._customer_id = customer_id
        self._items: List[OrderLine] = []
        self._status = OrderStatus.DRAFT
        self._created_at = datetime.now()
    
    @property
    def id(self) -> OrderId:
        return self._id
    
    def add_item(self, product_id: ProductId, quantity: int, price: Money):
        if self._status != OrderStatus.DRAFT:
            raise OrderNotEditableError()
        
        line = OrderLine(product_id, quantity, price)
        self._items.append(line)
    
    def submit(self):
        if not self._items:
            raise CannotSubmitEmptyOrderError()
        
        self._status = OrderStatus.SUBMITTED
        self._submitted_at = datetime.now()

# Immutable value object
@dataclass(frozen=True)
class Money:
    amount: Decimal
    currency: str
    
    def __post_init__(self):
        if self.amount < 0:
            raise MoneyCannotBeNegative()
    
    def __add__(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise CurrencyMismatch()
        return Money(self.amount + other.amount, self.currency)
    
    def __mul__(self, multiplier: Decimal) -> "Money":
        return Money(self.amount * multiplier, self.currency)

# Usage
total = Money("100.00", "USD") + Money("10.00", "USD")
# Returns Money(Decimal("110.00"), "USD")

# Aggregate root
class OrderAggregate:
    """Aggregate root that enforces invariants."""
    
    def __init__(self, order_id: OrderId):
        self._id = order_id
        self._items: List[OrderLine] = []
        self._status = OrderStatus.DRAFT
    
    # Command methods - the only way to modify state
    def add_item(self, product_id, quantity, price):
        # Invariant: Can't add items after submission
        if self._status != OrderStatus.DRAFT:
            raise OrderAlreadySubmittedError()
        
        # Invariant: Quantity must be positive
        if quantity <= 0:
            raise InvalidQuantityError()
        
        self._items.append(OrderLine(product_id, quantity, price))
    
    def submit(self):
        # Invariant: Can't submit empty order
        if not self._items:
            raise CannotSubmitEmptyOrderError()
        
        self._status = OrderStatus.SUBMITTED
    
    # Query methods - no side effects
    def get_total(self) -> Money:
        return sum((line.total for line in self._items), Money("0", "USD"))
    
    def get_status(self) -> OrderStatus:
        return self._status

# Domain event
@dataclass
class OrderSubmitted:
    order_id: OrderId
    customer_id: CustomerId
    total: Money
    occurred_at: datetime = field(default_factory=datetime.now)

# Event handler
class OrderEventHandler:
    def __init__(self, event_bus, inventory_service):
        self.event_bus = event_bus
        self.inventory_service = inventory_service
    
    def handle_order_submitted(self, event: OrderSubmitted):
        # Reserve inventory asynchronously
        asyncio.create_task(
            self.inventory_service.reserve(event.order_id, event.items)
        )

# Publishing events
class Order:
    def submit(self):
        self._status = OrderStatus.SUBMITTED
        
        # Publish domain event
        event = OrderSubmitted(
            order_id=self._id,
            customer_id=self._customer_id,
            total=self.get_total()
        )
        self.event_bus.publish(event)

# Shared glossary between technical and business teams
ubiquitous_language = {
    "Order": {
        "definition": "A customer's request to purchase products",
        "lifecycle": ["Draft", "Submitted", "Confirmed", "Shipped", "Delivered", "Cancelled"],
        "business_rules": [
            "Must have at least one item",
            "Customer must have valid payment method",
            "Items must be in stock"
        ]
    },
    "Payment": {
        "definition": "Authorization and capture of funds",
        "lifecycle": ["Pending", "Authorized", "Captured", "Failed", "Refunded"],
        "business_rules": [
            "Must be authorized before capturing",
            "Refund must be within 30 days"
        ]
    }
}

# DDD Layered Architecture
layers = {
    "presentation": {
        "responsibility": "HTTP handling, UI",
        "example": "FastAPI routes, React components"
    },
    "application": {
        "responsibility": "Use cases, orchestration",
        "example": "CreateOrderUseCase, ProcessPaymentUseCase"
    },
    "domain": {
        "responsibility": "Business logic, entities, rules",
        "example": "Order, Payment, Inventory entities"
    },
    "infrastructure": {
        "responsibility": "External concerns",
        "example": "Database repositories, payment gateway clients"
    }
}

# Repository interface (domain layer)
class OrderRepository(Protocol):
    def get_by_id(self, order_id: OrderId) -> Optional[Order]: ...
    def save(self, order: Order) -> None: ...
    def find_by_customer(self, customer_id: CustomerId) -> List[Order]: ...

# Repository implementation (infrastructure layer)
class PostgresOrderRepository(OrderRepository):
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def get_by_id(self, order_id: OrderId) -> Optional[Order]:
        result = await self.db.execute(
            select(OrderEntity).where(OrderEntity.id == order_id.value)
        )
        entity = result.scalar_one_or_none()
        return self._to_domain(entity) if entity else None
    
    async def save(self, order: Order) -> None:
        entity = self._to_entity(order)
        self.db.add(entity)
        await self.db.commit()