Cas avances¶

# Mauvais : vocabulaire technique, pas metier
class Data:
    def process(self): ...

class Manager:
    def handle(self, data): ...

# Bon : vocabulaire du domaine
class Loan:
    def approve(self, underwriter: Underwriter): ...
    def disburse(self, account: BankAccount): ...

class RiskAssessment:
    def evaluate(self, application: LoanApplication) -> RiskScore: ...

# Order est la racine d'agregat — OrderItem ne s'acces que via Order
class Order:
    def __init__(self, id: str, customer_id: str):
        self.id = id
        self.customer_id = customer_id
        self._items: list[OrderItem] = []
        self._version = 0

    def add_item(self, product_id: str, qty: int, price: float):
        if self.status != "draft":
            raise DomainError("Impossible d'ajouter un article a une commande confirmee")
        # invariant verifie a l'interieur de l'agregat
        self._items.append(OrderItem(product_id, qty, price))

    def confirm(self):
        if not self._items:
            raise DomainError("Commande vide")
        if self.total() <= 0:
            raise DomainError("Total invalide")
        self.status = "confirmed"
        return [OrderConfirmed(order_id=self.id, total=self.total())]

# Commands — modifient l'etat, retournent des evenements
@dataclass
class CreateOrderCommand:
    customer_id: str
    items: list[dict]

class CreateOrderHandler:
    def __init__(self, repo: OrderRepository, bus: EventBus):
        self._repo = repo
        self._bus = bus

    def handle(self, cmd: CreateOrderCommand) -> str:
        order = Order.create(cmd.customer_id, cmd.items)
        events = order.confirm()
        self._repo.save(order)
        for event in events:
            self._bus.publish(event)
        return order.id

# Queries — lisent l'etat, ne modifient rien
@dataclass
class GetOrderSummaryQuery:
    order_id: str

class GetOrderSummaryHandler:
    def __init__(self, read_db):
        self._db = read_db

    def handle(self, query: GetOrderSummaryQuery) -> dict:
        # requete directe sur une vue denormalisee — pas de domain objects
        return self._db.fetch_one(
            "SELECT * FROM order_summaries WHERE id = ?",
            (query.order_id,)
        )

# Sans event sourcing — on stocke l'etat final
orders table:
  id=1, status="shipped", total=99.90, updated_at=...

# Avec event sourcing — on stocke les evenements
events table:
  order_id=1, type="OrderCreated",   data={...}, ts=T1
  order_id=1, type="ItemAdded",      data={...}, ts=T2
  order_id=1, type="OrderConfirmed", data={...}, ts=T3
  order_id=1, type="OrderShipped",   data={...}, ts=T4

class Order:
    def __init__(self):
        self._events: list = []
        self.status = None
        self.items = []

    @classmethod
    def reconstitute(cls, events: list) -> "Order":
        order = cls()
        for event in events:
            order._apply(event)
        return order

    def _apply(self, event):
        handlers = {
            "OrderCreated": self._on_created,
            "ItemAdded": self._on_item_added,
            "OrderConfirmed": self._on_confirmed,
        }
        handlers[event["type"]](event["data"])

    def _on_created(self, data):
        self.id = data["id"]
        self.status = "draft"

    def _on_confirmed(self, data):
        self.status = "confirmed"

# Le legacy a un modele pollution dans votre nouveau code
# L'ACL traduit sans contaminer

class LegacyOrderAdapter:
    """Anti-Corruption Layer : isole notre domaine du modele legacy."""

    def __init__(self, legacy_client):
        self._legacy = legacy_client

    def get_order(self, order_id: str) -> Order:
        # le legacy retourne un dict avec des champs cryptiques
        raw = self._legacy.fetch_order_data(order_id)
        # on traduit vers notre modele propre
        return Order(
            id=raw["ORD_NUM"],
            customer_id=raw["CUST_REF"],
            status=self._map_status(raw["STAT_CODE"]),
        )

    def _map_status(self, legacy_code: str) -> str:
        mapping = {"01": "draft", "02": "confirmed", "05": "shipped"}
        return mapping.get(legacy_code, "unknown")