Domain-Driven Design: Building Domain-Centric Applications

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

class Entity:
    """Base entity with identity."""
    
    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)

@dataclass(frozen=True)
class Address:
    """Value object - immutable and defined by values."""
    street: str
    city: str
    state: str
    zip_code: str
    country: str

class User(Entity):
    """Entity with business logic."""
    
    def __init__(self, email: str, name: str):
        super().__init__()
        self._email = email
        self._name = name
        self._created_at = datetime.utcnow()
        self._is_active = True
    
    @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 activate(self):
        self._is_active = True
    
    def deactivate(self):
        self._is_active = False
    
    def change_email(self, new_email: str):
        if "@" not in new_email:
            raise ValueError("Invalid email")
        self._email = new_email

class OrderItem:
    """Part of an aggregate."""
    
    def __init__(self, product_id: str, quantity: int, unit_price: float):
        self.product_id = product_id
        self.quantity = quantity
        self.unit_price = unit_price
    
    @property
    def total(self) -> float:
        return self.quantity * self.unit_price

class Order(AggregateRoot):
    """Aggregate root - controls access to aggregate."""
    
    def __init__(self, user_id: str):
        super().__init__()
        self._user_id = user_id
        self._items: List[OrderItem] = []
        self._status = "pending"
    
    def add_item(self, product_id: str, quantity: int, unit_price: float):
        item = OrderItem(product_id, quantity, unit_price)
        self._items.append(item)
    
    def submit(self):
        if not self._items:
            raise ValueError("Cannot submit empty order")
        self._status = "submitted"
    
    def cancel(self):
        if self._status == "shipped":
            raise ValueError("Cannot cancel shipped order")
        self._status = "cancelled"
    
    @property
    def total(self) -> float:
        return sum(item.total for item in self._items)

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