15. DDD tactique

15. DDD tactique#

Si le DDD stratégique répond à la question « comment décomposer le système en grandes parties cohérentes ? », le DDD tactique répond à la question « comment organiser le code à l’intérieur d’un Bounded Context ? ». Les outils tactiques — entités, value objects, agrégats, repositories, domain events, domain services, application services et factories — forment un vocabulaire de conception précis qui permet d’exprimer les règles métier de façon explicite et vérifiable.

Entités — identité et cycle de vie#

Une entité est un objet dont l’identité est ce qui le définit, indépendamment de ses attributs. Deux entités avec les mêmes attributs mais des identifiants différents sont des objets distincts.

Caractéristiques d’une entité#

Une entité possède :

  • Un identifiant stable qui ne change pas au fil du temps (UUID, ID métier)

  • Un cycle de vie : elle est créée, modifiée, parfois désactivée ou supprimée

  • De la mutabilité : ses attributs évoluent, son identité reste fixe

  • Des invariants : des règles qui doivent toujours être vraies

# Entité Order — identité par order_id
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import List
import uuid

class OrderStatus(Enum):
    PENDING    = auto()
    CONFIRMED  = auto()
    SHIPPED    = auto()
    DELIVERED  = auto()
    CANCELLED  = auto()

@dataclass
class Customer:
    customer_id: str
    name: str
    email: str

    def __eq__(self, other):
        if not isinstance(other, Customer):
            return False
        return self.customer_id == other.customer_id  # identité par ID

    def __hash__(self):
        return hash(self.customer_id)

Quand utiliser une entité#

Utiliser une entité quand l’objet a une continuité dans le temps et qu’on a besoin de le retrouver, le modifier ou le suivre. Exemples : Order, Customer, Product, BankAccount, Employee.

Identité métier vs identité technique

Préférer un identifiant métier stable (numéro de commande, IBAN) à un identifiant technique auto-incrémenté quand il existe. L’identifiant technique change si la base de données est migrée ; l’identifiant métier est stable dans le temps et compréhensible par les experts métier.

Value Objects — immutabilité et égalité par valeur#

Un Value Object est un objet défini par ses attributs, sans identité propre. Deux value objects avec les mêmes attributs sont interchangeables. Ils sont immuables : pour « modifier » un value object, on en crée un nouveau.

Avantages#

  • Sécurité : l’immuabilité élimine les bugs de partage d’état

  • Expressivité : Money(100, "EUR") est plus expressif que float(100)

  • Validation : les règles de validité sont dans le constructeur, garantissant qu’un value object invalide ne peut pas exister

  • Testabilité : les fonctions sur des value objects sont pures

Exemples Python complets#

from __future__ import annotations
from dataclasses import dataclass
from typing import ClassVar
import re

# --- Value Object : Money ---

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str

    SUPPORTED_CURRENCIES: ClassVar[set] = {"EUR", "USD", "GBP", "CHF"}

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError(f"Le montant ne peut pas être négatif : {self.amount}")
        if self.currency not in self.SUPPORTED_CURRENCIES:
            raise ValueError(f"Devise non supportée : {self.currency}")
        # Arrondir à 2 décimales (frozen=True → utiliser object.__setattr__)
        object.__setattr__(self, "amount", round(self.amount, 2))

    def __add__(self, other: Money) -> Money:
        if self.currency != other.currency:
            raise ValueError(
                f"Impossible d'additionner {self.currency} et {other.currency}"
            )
        return Money(self.amount + other.amount, self.currency)

    def __sub__(self, other: Money) -> Money:
        if self.currency != other.currency:
            raise ValueError(
                f"Impossible de soustraire {self.currency} et {other.currency}"
            )
        result = self.amount - other.amount
        if result < 0:
            raise ValueError("Le résultat serait négatif")
        return Money(result, self.currency)

    def __mul__(self, factor: float) -> Money:
        return Money(self.amount * factor, self.currency)

    def __le__(self, other: Money) -> bool:
        self._check_same_currency(other)
        return self.amount <= other.amount

    def __lt__(self, other: Money) -> bool:
        self._check_same_currency(other)
        return self.amount < other.amount

    def _check_same_currency(self, other: Money) -> None:
        if self.currency != other.currency:
            raise ValueError("Comparaison entre devises différentes")

    def __repr__(self) -> str:
        return f"{self.amount:.2f} {self.currency}"


# --- Value Object : Email ---

@dataclass(frozen=True)
class Email:
    value: str

    _PATTERN: ClassVar[re.Pattern] = re.compile(
        r"^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$"
    )

    def __post_init__(self):
        normalized = self.value.strip().lower()
        if not self._PATTERN.match(normalized):
            raise ValueError(f"Email invalide : {self.value!r}")
        object.__setattr__(self, "value", normalized)

    @property
    def domain(self) -> str:
        return self.value.split("@")[1]

    def __repr__(self) -> str:
        return self.value


# --- Value Object : Address ---

@dataclass(frozen=True)
class Address:
    street: str
    city: str
    postal_code: str
    country: str

    def __post_init__(self):
        for field_name, val in [
            ("street", self.street),
            ("city", self.city),
            ("postal_code", self.postal_code),
            ("country", self.country),
        ]:
            if not val or not val.strip():
                raise ValueError(f"Le champ '{field_name}' ne peut pas être vide")

    def with_city(self, new_city: str) -> Address:
        """Retourne une nouvelle adresse avec la ville modifiée (immuabilité)."""
        return Address(self.street, new_city, self.postal_code, self.country)

    def __repr__(self) -> str:
        return f"{self.street}, {self.postal_code} {self.city}, {self.country}"


# --- Tests intégrés ---

print("=== Tests Value Object : Money ===")
price  = Money(19.99, "EUR")
tax    = Money(2.00,  "EUR")
total  = price + tax
print(f"  {price} + {tax} = {total}")
print(f"  {total} × 2 = {total * 2}")
print(f"  Immuable : price après opérations = {price}")

try:
    Money(-5, "EUR")
except ValueError as e:
    print(f"  Validation OK : {e}")

try:
    Money(10, "EUR") + Money(10, "USD")
except ValueError as e:
    print(f"  Validation OK : {e}")

print()
print("=== Tests Value Object : Email ===")
email = Email("  Alice@Example.COM  ")
print(f"  Normalisé : {email}")
print(f"  Domaine   : {email.domain}")
try:
    Email("pas-un-email")
except ValueError as e:
    print(f"  Validation OK : {e}")

print()
print("=== Tests Value Object : Address ===")
addr  = Address("12 rue de la Paix", "Paris", "75001", "France")
addr2 = addr.with_city("Lyon")
print(f"  Original  : {addr}")
print(f"  Modifiée  : {addr2}")
print(f"  Égalité   : addr == addr  → {addr == addr}")
print(f"  Égalité   : addr == addr2 → {addr == addr2}")

=== Tests Value Object : Money ===
  19.99 EUR + 2.00 EUR = 21.99 EUR
  21.99 EUR × 2 = 43.98 EUR
  Immuable : price après opérations = 19.99 EUR
  Validation OK : Le montant ne peut pas être négatif : -5
  Validation OK : Impossible d'additionner EUR et USD

=== Tests Value Object : Email ===
  Normalisé : alice@example.com
  Domaine   : example.com
  Validation OK : Email invalide : 'pas-un-email'

=== Tests Value Object : Address ===
  Original  : 12 rue de la Paix, 75001 Paris, France
  Modifiée  : 12 rue de la Paix, 75001 Lyon, France
  Égalité   : addr == addr  → True
  Égalité   : addr == addr2 → False

frozen=True et post_init

Avec @dataclass(frozen=True), les attributs ne peuvent plus être modifiés après construction. Pour normaliser une valeur dans __post_init__ (mise en minuscules d’un email, arrondi d’un montant), utiliser object.__setattr__(self, "attr", valeur) — c’est le seul moyen de contourner la protection de façon explicite et intentionnelle.

Agrégats — frontière de cohérence transactionnelle#

Un agrégat est un cluster d’entités et de value objects traité comme une unité pour les modifications. Il garantit la cohérence transactionnelle à l’intérieur de ses frontières.

L’Aggregate Root#

Chaque agrégat a une racine (aggregate root) : l’entité centrale par laquelle toutes les interactions extérieures passent. Personne ne peut accéder directement aux objets internes de l’agrégat — uniquement via la racine.

Règles fondamentales :

  1. Les invariants de l’agrégat doivent être vérifiés à chaque modification

  2. Seule la racine est référencée de l’extérieur (par son ID)

  3. Une transaction ne modifie qu’un seul agrégat (règle de base — exceptions possibles avec la saga)

  4. Les autres agrégats sont référencés par leur ID, jamais par leur objet

Taille des agrégats#

Un agrégat trop grand est un antipattern fréquent. Si un Order contient les données du client, les articles, les détails de paiement et les informations de livraison, chaque opération sur cet agrégat verrouille l’ensemble. La règle pratique : un agrégat doit contenir le minimum nécessaire pour garantir ses invariants métier.

Implémentation complète : agrégat Order#

# Agrégat Order — implémentation complète
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import List
import uuid

class OrderStatus(Enum):
    PENDING    = auto()
    CONFIRMED  = auto()
    SHIPPED    = auto()
    DELIVERED  = auto()
    CANCELLED  = auto()

# --- Domain Events ---

@dataclass(frozen=True)
class DomainEvent:
    occurred_at: datetime = field(
        default_factory=datetime.utcnow
    )

@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str = ""
    customer_id: str = ""
    total: float = 0.0

@dataclass(frozen=True)
class ItemAdded(DomainEvent):
    order_id: str = ""
    product_id: str = ""
    quantity: int = 0
    unit_price: float = 0.0

@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
    order_id: str = ""
    reason: str = ""

# --- Value Objects ---

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str = "EUR"

    def __add__(self, other: Money) -> Money:
        assert self.currency == other.currency
        return Money(round(self.amount + other.amount, 2), self.currency)

    def __mul__(self, factor: float) -> Money:
        return Money(round(self.amount * factor, 2), self.currency)

    def __repr__(self):
        return f"{self.amount:.2f} {self.currency}"

# --- Entité interne (OrderItem) ---

@dataclass
class OrderItem:
    product_id: str
    product_name: str
    unit_price: Money
    quantity: int

    def __post_init__(self):
        if self.quantity <= 0:
            raise ValueError("La quantité doit être positive")

    @property
    def subtotal(self) -> Money:
        return self.unit_price * self.quantity

# --- Aggregate Root ---

class Order:
    MAX_ITEMS = 50

    def __init__(self, order_id: str, customer_id: str):
        self._id = order_id
        self._customer_id = customer_id
        self._items: List[OrderItem] = []
        self._status = OrderStatus.PENDING
        self._events: List[DomainEvent] = []
        self._version = 0

    # ---- Comportements métier ----

    def add_item(self, product_id: str, product_name: str,
                 unit_price: Money, quantity: int) -> None:
        self._assert_status(OrderStatus.PENDING,
                            "Impossible d'ajouter un article")
        if len(self._items) >= self.MAX_ITEMS:
            raise ValueError(
                f"Une commande ne peut pas dépasser {self.MAX_ITEMS} articles"
            )
        # Fusionner si le produit existe déjà
        for item in self._items:
            if item.product_id == product_id:
                item.quantity += quantity
                self._record(ItemAdded(
                    order_id=self._id,
                    product_id=product_id,
                    quantity=quantity,
                    unit_price=unit_price.amount
                ))
                return
        self._items.append(
            OrderItem(product_id, product_name, unit_price, quantity)
        )
        self._record(ItemAdded(
            order_id=self._id,
            product_id=product_id,
            quantity=quantity,
            unit_price=unit_price.amount
        ))

    def remove_item(self, product_id: str) -> None:
        self._assert_status(OrderStatus.PENDING,
                            "Impossible de retirer un article")
        before = len(self._items)
        self._items = [i for i in self._items
                       if i.product_id != product_id]
        if len(self._items) == before:
            raise ValueError(
                f"Produit {product_id!r} introuvable dans la commande"
            )

    def place(self) -> None:
        """Confirme la commande — vérifie les invariants."""
        self._assert_status(OrderStatus.PENDING, "Commande déjà traitée")
        if not self._items:
            raise ValueError("Impossible de confirmer une commande vide")
        self._status = OrderStatus.CONFIRMED
        self._record(OrderPlaced(
            order_id=self._id,
            customer_id=self._customer_id,
            total=self.total.amount
        ))

    def cancel(self, reason: str) -> None:
        if self._status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
            raise ValueError(
                "Impossible d'annuler une commande déjà expédiée ou livrée"
            )
        self._status = OrderStatus.CANCELLED
        self._record(OrderCancelled(order_id=self._id, reason=reason))

    # ---- Propriétés ----

    @property
    def id(self) -> str:
        return self._id

    @property
    def status(self) -> OrderStatus:
        return self._status

    @property
    def total(self) -> Money:
        if not self._items:
            return Money(0.0)
        result = self._items[0].subtotal
        for item in self._items[1:]:
            result = result + item.subtotal
        return result

    @property
    def item_count(self) -> int:
        return sum(i.quantity for i in self._items)

    # ---- Gestion des événements ----

    def _record(self, event: DomainEvent) -> None:
        self._events.append(event)

    def pull_events(self) -> List[DomainEvent]:
        events, self._events = self._events, []
        return events

    # ---- Invariants ----

    def _assert_status(self, expected: OrderStatus, msg: str) -> None:
        if self._status != expected:
            raise ValueError(
                f"{msg} : statut actuel = {self._status.name}, "
                f"attendu = {expected.name}"
            )

    def __repr__(self) -> str:
        return (f"Order(id={self._id!r}, status={self._status.name}, "
                f"total={self.total}, items={len(self._items)})")

Agrégat Order — cellule exécutable#

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import List
import uuid

class OrderStatus(Enum):
    PENDING    = auto()
    CONFIRMED  = auto()
    SHIPPED    = auto()
    DELIVERED  = auto()
    CANCELLED  = auto()

@dataclass(frozen=True)
class Money:
    amount: float
    currency: str = "EUR"
    def __add__(self, other):
        return Money(round(self.amount + other.amount, 2), self.currency)
    def __mul__(self, f):
        return Money(round(self.amount * f, 2), self.currency)
    def __repr__(self):
        return f"{self.amount:.2f} {self.currency}"

@dataclass(frozen=True)
class DomainEvent:
    occurred_at: datetime = field(default_factory=datetime.utcnow)

@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str = ""
    customer_id: str = ""
    total: float = 0.0

@dataclass(frozen=True)
class ItemAdded(DomainEvent):
    order_id: str = ""
    product_id: str = ""
    quantity: int = 0

@dataclass(frozen=True)
class OrderCancelled(DomainEvent):
    order_id: str = ""
    reason: str = ""

@dataclass
class OrderItem:
    product_id: str
    product_name: str
    unit_price: Money
    quantity: int
    def __post_init__(self):
        if self.quantity <= 0:
            raise ValueError("Quantité invalide")
    @property
    def subtotal(self):
        return self.unit_price * self.quantity

class Order:
    MAX_ITEMS = 50

    def __init__(self, order_id: str, customer_id: str):
        self._id = order_id
        self._customer_id = customer_id
        self._items: List[OrderItem] = []
        self._status = OrderStatus.PENDING
        self._events: List[DomainEvent] = []

    def add_item(self, product_id, product_name, unit_price, quantity):
        if self._status != OrderStatus.PENDING:
            raise ValueError("Commande déjà confirmée")
        if len(self._items) >= self.MAX_ITEMS:
            raise ValueError("Trop d'articles")
        for item in self._items:
            if item.product_id == product_id:
                item.quantity += quantity
                self._events.append(ItemAdded(order_id=self._id,
                    product_id=product_id, quantity=quantity))
                return
        self._items.append(OrderItem(product_id, product_name,
                                     unit_price, quantity))
        self._events.append(ItemAdded(order_id=self._id,
            product_id=product_id, quantity=quantity))

    def place(self):
        if self._status != OrderStatus.PENDING:
            raise ValueError("Commande déjà traitée")
        if not self._items:
            raise ValueError("Commande vide")
        self._status = OrderStatus.CONFIRMED
        self._events.append(OrderPlaced(order_id=self._id,
            customer_id=self._customer_id, total=self.total.amount))

    def cancel(self, reason):
        if self._status in (OrderStatus.SHIPPED, OrderStatus.DELIVERED):
            raise ValueError("Impossible d'annuler")
        self._status = OrderStatus.CANCELLED
        self._events.append(OrderCancelled(order_id=self._id, reason=reason))

    @property
    def total(self):
        if not self._items:
            return Money(0.0)
        result = self._items[0].subtotal
        for item in self._items[1:]:
            result = result + item.subtotal
        return result

    def pull_events(self):
        events, self._events = self._events, []
        return events

    def __repr__(self):
        return (f"Order({self._id[:8]}… | {self._status.name} | "
                f"{self.total} | {len(self._items)} ligne(s))")


# --- Démonstration ---
order_id = str(uuid.uuid4())
order = Order(order_id, customer_id="CUST-001")

order.add_item("PROD-A", "Clavier mécanique", Money(89.99), 1)
order.add_item("PROD-B", "Souris ergonomique", Money(45.00), 2)
order.add_item("PROD-A", "Clavier mécanique", Money(89.99), 1)  # fusion

print("Après ajout d'articles :", order)
print()

events = order.pull_events()
print(f"Événements émis ({len(events)}) :")
for e in events:
    print(f"  [{type(e).__name__}] {e}")

print()
order.place()
print("Après confirmation :", order)

events = order.pull_events()
print(f"Événements émis ({len(events)}) :")
for e in events:
    print(f"  [{type(e).__name__}] {e}")

print()
# Test des invariants
try:
    order.add_item("PROD-C", "Écran", Money(299.0), 1)
except ValueError as e:
    print(f"Invariant respecté : {e}")

try:
    order2 = Order(str(uuid.uuid4()), "CUST-002")
    order2.place()
except ValueError as e:
    print(f"Invariant respecté : {e}")

Après ajout d'articles : Order(4da78569… | PENDING | 269.98 EUR | 2 ligne(s))

Événements émis (3) :
  [ItemAdded] ItemAdded(occurred_at=datetime.datetime(2026, 3, 25, 0, 28, 24, 429018), order_id='4da78569-6894-47a7-8967-b1f7a1124a54', product_id='PROD-A', quantity=1)
  [ItemAdded] ItemAdded(occurred_at=datetime.datetime(2026, 3, 25, 0, 28, 24, 429096), order_id='4da78569-6894-47a7-8967-b1f7a1124a54', product_id='PROD-B', quantity=2)
  [ItemAdded] ItemAdded(occurred_at=datetime.datetime(2026, 3, 25, 0, 28, 24, 429142), order_id='4da78569-6894-47a7-8967-b1f7a1124a54', product_id='PROD-A', quantity=1)

Après confirmation : Order(4da78569… | CONFIRMED | 269.98 EUR | 2 ligne(s))
Événements émis (1) :
  [OrderPlaced] OrderPlaced(occurred_at=datetime.datetime(2026, 3, 25, 0, 28, 24, 429623), order_id='4da78569-6894-47a7-8967-b1f7a1124a54', customer_id='CUST-001', total=269.98)

Invariant respecté : Commande déjà confirmée
Invariant respecté : Commande vide

<string>:3: DeprecationWarning: datetime.datetime.utcnow() is deprecated and scheduled for removal in a future version. Use timezone-aware objects to represent datetimes in UTC: datetime.datetime.now(datetime.UTC).

Repositories — accès aux agrégats#

Un repository abstrait le mécanisme de persistance. Il fournit une interface qui ressemble à une collection en mémoire, cachant les détails de la base de données.

Séparation interface / implémentation#

L’interface du repository appartient à la couche domaine — elle exprime ce dont le domaine a besoin. L’implémentation appartient à la couche infrastructure — elle traduit ces besoins en SQL, appels d’API, etc.

# Interface dans le domaine (couche domain)
from abc import ABC, abstractmethod
from typing import Optional, List

class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, order_id: str) -> Optional[Order]:
        """Retourne None si la commande n'existe pas."""
        ...

    @abstractmethod
    def find_by_customer(self, customer_id: str) -> List[Order]:
        """Retourne toutes les commandes d'un client."""
        ...

    @abstractmethod
    def save(self, order: Order) -> None:
        """Persiste un agrégat (insertion ou mise à jour)."""
        ...

    @abstractmethod
    def delete(self, order_id: str) -> None:
        ...


# Implémentation en mémoire (tests, prototypage)
class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self._store: dict[str, Order] = {}

    def find_by_id(self, order_id: str) -> Optional[Order]:
        return self._store.get(order_id)

    def find_by_customer(self, customer_id: str) -> List[Order]:
        return [o for o in self._store.values()
                if o._customer_id == customer_id]

    def save(self, order: Order) -> None:
        self._store[order.id] = order

    def delete(self, order_id: str) -> None:
        self._store.pop(order_id, None)


# Implémentation SQL (production)
class SqlOrderRepository(OrderRepository):
    def __init__(self, session):
        self._session = session

    def find_by_id(self, order_id: str) -> Optional[Order]:
        row = self._session.execute(
            "SELECT * FROM orders WHERE id = :id",
            {"id": order_id}
        ).fetchone()
        return self._reconstruct(row) if row else None

    def save(self, order: Order) -> None:
        # Mapping domaine → base de données
        self._session.execute(
            """INSERT INTO orders (id, customer_id, status, total)
               VALUES (:id, :customer_id, :status, :total)
               ON CONFLICT (id) DO UPDATE
               SET status = :status, total = :total""",
            {
                "id": order.id,
                "customer_id": order._customer_id,
                "status": order.status.name,
                "total": order.total.amount,
            }
        )

    def _reconstruct(self, row) -> Order:
        # Reconstruction de l'agrégat depuis les données brutes
        order = Order.__new__(Order)
        order._id = row["id"]
        order._customer_id = row["customer_id"]
        order._status = OrderStatus[row["status"]]
        order._items = []
        order._events = []
        return order

    def find_by_customer(self, customer_id: str) -> List[Order]:
        rows = self._session.execute(
            "SELECT * FROM orders WHERE customer_id = :cid",
            {"cid": customer_id}
        ).fetchall()
        return [self._reconstruct(r) for r in rows]

    def delete(self, order_id: str) -> None:
        self._session.execute(
            "DELETE FROM orders WHERE id = :id", {"id": order_id}
        )

Un repository par agrégat

Un repository n’expose que des agrégats complets, jamais des entités internes. On ne crée pas un OrderItemRepository — les OrderItem sont toujours accédés via OrderRepository. Cette règle garantit que les invariants de l’agrégat sont toujours vérifiés.

Domain Events — ce qui s’est passé#

Un Domain Event représente quelque chose qui s’est produit dans le domaine et qui est significatif pour les experts métier. Le nommage au passé est fondamental : OrderPlaced, PaymentProcessed, ItemShipped — pas PlaceOrder (qui est une commande).

Rôle des domain events#

  • Découplage entre Bounded Contexts : le contexte Shipping s’abonne à OrderPlaced sans connaître le contexte Order

  • Traçabilité : l’historique des événements est une source de vérité

  • Event Sourcing : reconstruire l’état d’un agrégat à partir de ses événements

  • Saga / Process Manager : orchestrer des workflows multi-agrégats

# Domain Events bien nommés
@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
    order_id: str
    customer_id: str
    items: list
    total_amount: float
    placed_at: datetime

@dataclass(frozen=True)
class PaymentProcessed(DomainEvent):
    order_id: str
    payment_id: str
    amount: float
    provider: str
    processed_at: datetime

@dataclass(frozen=True)
class ItemShipped(DomainEvent):
    order_id: str
    shipment_id: str
    carrier: str
    tracking_number: str
    shipped_at: datetime

@dataclass(frozen=True)
class OrderDelivered(DomainEvent):
    order_id: str
    delivered_at: datetime

# Mauvais nommage — à éviter
# class ShipOrder(DomainEvent): ...        # commande, pas événement
# class OrderUpdated(DomainEvent): ...     # trop vague
# class OrderStatusChanged(DomainEvent):  # technique, pas métier

Domain Services — logique inter-entités#

Un Domain Service contient de la logique métier qui n’appartient naturellement à aucune entité ou value object particulier — typiquement une opération qui implique plusieurs entités.

# Domain Service : TransferService
class TransferService:
    """
    Orchestre un transfert de fonds entre deux comptes.
    Cette logique n'appartient ni à SourceAccount ni à TargetAccount —
    elle concerne les deux simultanément.
    """
    def __init__(self, event_publisher):
        self._publisher = event_publisher

    def transfer(self, source: BankAccount, target: BankAccount,
                 amount: Money) -> None:
        if source.currency != target.currency:
            raise DomainException(
                "Transfert entre comptes de devises différentes "
                "non supporté sans conversion explicite"
            )
        source.debit(amount)   # vérifie les invariants du compte source
        target.credit(amount)  # vérifie les invariants du compte cible
        self._publisher.publish(
            FundsTransferred(
                source_id=source.id,
                target_id=target.id,
                amount=amount.amount,
                currency=amount.currency
            )
        )


# Domain Service : PricingService
class PricingService:
    """
    Calcule le prix final d'un article en tenant compte
    des promotions, de la segmentation client et des règles tarifaires.
    Cette logique est trop complexe pour vivre dans Product ou Customer.
    """
    def __init__(self, promotion_repo, pricing_rules):
        self._promotions = promotion_repo
        self._rules = pricing_rules

    def calculate_price(self, product: Product,
                        customer: Customer,
                        quantity: int) -> Money:
        base_price = product.base_price * quantity
        applicable_promotions = self._promotions.find_for(
            product, customer, quantity
        )
        return self._rules.apply(base_price, applicable_promotions,
                                 customer.segment)

Domain Service vs Application Service

Un Domain Service contient de la logique métier (règles du domaine). Un Application Service orchestre un cas d’usage (section suivante). Ne pas confondre les deux : un Domain Service ne connaît pas les transactions, la sécurité ou le transport réseau.

Application Services — orchestration des use cases#

Un Application Service est le point d’entrée d’un cas d’usage. Il orchestre le flux : récupérer les agrégats depuis les repositories, appeler les méthodes métier, persister, publier les événements. Il ne contient aucune logique métier.

# Application Service : PlaceOrderCommandHandler
@dataclass
class PlaceOrderCommand:
    customer_id: str
    items: list  # [{"product_id": str, "quantity": int}]
    payment_method_id: str

class PlaceOrderHandler:
    def __init__(self, order_repo: OrderRepository,
                 product_repo: ProductRepository,
                 pricing_service: PricingService,
                 event_publisher):
        self._orders = order_repo
        self._products = product_repo
        self._pricing = pricing_service
        self._publisher = event_publisher

    def handle(self, cmd: PlaceOrderCommand) -> str:
        # 1. Récupérer les données nécessaires
        customer = self._orders.find_customer(cmd.customer_id)

        # 2. Créer l'agrégat (via Factory implicite)
        order = Order(
            order_id=str(uuid.uuid4()),
            customer_id=cmd.customer_id
        )

        # 3. Appliquer la logique métier (dans l'agrégat)
        for item_dto in cmd.items:
            product = self._products.find_by_id(item_dto["product_id"])
            price = self._pricing.calculate_price(
                product, customer, item_dto["quantity"]
            )
            order.add_item(
                product.id, product.name, price, item_dto["quantity"]
            )

        order.place()  # vérifie les invariants et émet l'événement

        # 4. Persister
        self._orders.save(order)

        # 5. Publier les événements de domaine
        for event in order.pull_events():
            self._publisher.publish(event)

        return order.id

Factory — création complexe d’agrégats#

Une Factory encapsule la logique de création d’un agrégat quand cette création est suffisamment complexe pour ne pas tenir dans un constructeur.

# Factory : OrderFactory
class OrderFactory:
    def __init__(self, pricing_service: PricingService,
                 id_generator):
        self._pricing = pricing_service
        self._id_gen = id_generator

    def create_from_cart(self, cart: Cart,
                         customer: Customer) -> Order:
        if not cart.items:
            raise DomainException("Impossible de créer une commande depuis un panier vide")

        order = Order(
            order_id=self._id_gen.generate(),
            customer_id=customer.id
        )
        for cart_item in cart.items:
            price = self._pricing.calculate_price(
                cart_item.product, customer, cart_item.quantity
            )
            order.add_item(
                cart_item.product.id,
                cart_item.product.name,
                price,
                cart_item.quantity
            )
        return order

    def reconstitute(self, snapshot: dict) -> Order:
        """Reconstruit un agrégat depuis un snapshot persisté."""
        order = Order.__new__(Order)
        order._id = snapshot["id"]
        order._customer_id = snapshot["customer_id"]
        order._status = OrderStatus[snapshot["status"]]
        order._items = [
            OrderItem(
                i["product_id"], i["product_name"],
                Money(i["unit_price"]), i["quantity"]
            )
            for i in snapshot["items"]
        ]
        order._events = []
        return order

Visualisations#

Diagramme des blocs tactiques DDD#

import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import seaborn as sns

sns.set_theme(style="whitegrid", palette="muted", font_scale=1.1)

fig, ax = plt.subplots(figsize=(14, 9))
ax.set_facecolor("#F5F7FA")
ax.set_xlim(0, 1)
ax.set_ylim(0, 1)

# Définition des blocs : (nom, x, y, largeur, hauteur, couleur, desc)
blocs = [
    # Couche Application
    ("Application\nService",  0.05, 0.78, 0.20, 0.12, "#4C72B0",
     "Orchestration\nuse cases"),
    ("Command /\nQuery",      0.30, 0.78, 0.17, 0.12, "#4C72B0",
     "DTO d'entrée"),
    ("Event\nPublisher",      0.52, 0.78, 0.17, 0.12, "#4C72B0",
     "Dispatch events"),
    # Couche Domaine
    ("Aggregate\nRoot",       0.05, 0.50, 0.22, 0.14, "#C44E52",
     "Invariants + état"),
    ("Entity",                0.32, 0.50, 0.14, 0.14, "#C44E52",
     "Identité + cycle vie"),
    ("Value\nObject",         0.50, 0.50, 0.14, 0.14, "#DD8452",
     "Immuable, égalité valeur"),
    ("Domain\nEvent",         0.68, 0.50, 0.14, 0.14, "#8172B2",
     "Ce qui s'est passé"),
    ("Domain\nService",       0.86, 0.50, 0.12, 0.14, "#55A868",
     "Logique inter-entités"),
    ("Repository\n(interface)",0.05, 0.28, 0.22, 0.12, "#937860",
     "Abstraction persistance"),
    ("Factory",               0.32, 0.28, 0.14, 0.12, "#937860",
     "Création complexe"),
    # Couche Infrastructure
    ("Repository\n(impl.)",   0.05, 0.08, 0.22, 0.12, "#6AAB9C",
     "SQL / NoSQL / API"),
    ("ORM /\nMapper",         0.32, 0.08, 0.14, 0.12, "#6AAB9C",
     "Persistance technique"),
    ("Message\nBroker",       0.52, 0.08, 0.17, 0.12, "#6AAB9C",
     "Kafka / RabbitMQ"),
]

for nom, x, y, w, h, color, desc in blocs:
    box = mpatches.FancyBboxPatch(
        (x, y), w, h,
        boxstyle="round,pad=0.01",
        facecolor=color, edgecolor="white",
        linewidth=2, zorder=3
    )
    ax.add_patch(box)
    ax.text(x + w/2, y + h*0.62, nom,
            ha="center", va="center",
            fontsize=8, fontweight="bold", color="white", zorder=4)
    ax.text(x + w/2, y + h*0.22, desc,
            ha="center", va="center",
            fontsize=6.5, color="white", alpha=0.88, zorder=4)

# Étiquettes de couche
for label, yc, color in [
    ("COUCHE APPLICATION",   0.90, "#4C72B0"),
    ("COUCHE DOMAINE",        0.68, "#C44E52"),
    ("COUCHE INFRASTRUCTURE", 0.22, "#6AAB9C"),
]:
    ax.axhline(y=yc - 0.02, xmin=0.01, xmax=0.99,
               color=color, linestyle="--", linewidth=1, alpha=0.4)
    ax.text(0.99, yc + 0.01, label,
            ha="right", va="bottom", fontsize=8,
            color=color, fontweight="bold", alpha=0.7)

# Flèches de dépendance
arrows = [
    (0.15, 0.78, 0.15, 0.64),   # AppService → AggregateRoot
    (0.15, 0.50, 0.15, 0.40),   # AggregateRoot → Repository (interface)
    (0.15, 0.28, 0.15, 0.20),   # Repository interface → impl
    (0.39, 0.50, 0.57, 0.50),   # Entity → ValueObject
    (0.64, 0.50, 0.77, 0.57),   # ValueObject → DomainEvent (diag)
    (0.75, 0.50, 0.61, 0.14),   # DomainEvent → MessageBroker
]
for x1, y1, x2, y2 in arrows:
    ax.annotate(
        "", xy=(x2, y2), xytext=(x1, y1),
        arrowprops=dict(arrowstyle="-|>", color="#555555",
                        lw=1.2, connectionstyle="arc3,rad=0.0"),
        zorder=2
    )

ax.set_title("Blocs tactiques DDD et leurs relations",
             fontsize=13, fontweight="bold", pad=10)
ax.axis("off")
plt.savefig("ddd_tactique_blocs.png", dpi=120,
            bbox_inches="tight", facecolor="#F5F7FA")
plt.show()
_images/533575fa8c72643d22e537071d21ebe2c777eeb14ddef80354b24aff89fa2ac6.png

Comparaison modèle anémique vs modèle riche#

import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
import seaborn as sns
import numpy as np

sns.set_theme(style="whitegrid", palette="muted", font_scale=1.1)

criteres = [
    "Localisation\nde la logique",
    "Garantie des\ninvariants",
    "Testabilité\nunitaire",
    "Lisibilité\ndu domaine",
    "Découverte\ndes règles",
    "Refactoring\nSécurisé",
    "Alignement\nlangage commun",
]

anemique = [2, 2, 3, 2, 2, 2, 2]
riche    = [5, 5, 5, 5, 4, 5, 5]

x = np.arange(len(criteres))
width = 0.35

fig, ax = plt.subplots(figsize=(13, 6))

bars1 = ax.bar(x - width/2, anemique, width,
               label="Modèle anémique", color="#C44E52", alpha=0.8,
               edgecolor="white", linewidth=1.5)
bars2 = ax.bar(x + width/2, riche, width,
               label="Modèle riche (DDD)", color="#55A868", alpha=0.8,
               edgecolor="white", linewidth=1.5)

# Annotations texte sur les barres
for bar, val in zip(bars1, anemique):
    ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.05,
            ["Très faible", "Très faible", "Faible", "Très faible",
             "Très faible", "Faible", "Très faible"][anemique.index(val)
             if val != 3 else 2],
            ha="center", va="bottom", fontsize=6.5, color="#C44E52",
            rotation=30)

for bar, val in zip(bars2, riche):
    label = "Excellent" if val == 5 else "Bon"
    ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.05,
            label, ha="center", va="bottom", fontsize=6.5,
            color="#2A7A4E", rotation=30)

ax.set_xticks(x)
ax.set_xticklabels(criteres, fontsize=9)
ax.set_yticks([1, 2, 3, 4, 5])
ax.set_yticklabels(["1\nTrès faible", "2\nFaible", "3\nMoyen",
                    "4\nBon", "5\nExcellent"], fontsize=8)
ax.set_ylim(0, 6.2)
ax.set_ylabel("Score (1–5)", fontsize=10)
ax.set_title("Modèle anémique vs modèle riche — comparaison par critère",
             fontsize=13, fontweight="bold")
ax.legend(fontsize=10)
plt.savefig("anemique_vs_riche.png", dpi=120,
            bbox_inches="tight", facecolor="white")
plt.show()
_images/1eb94c8b5906789c12bc99cd80bfe6ba2d083a5c9699c7aedd530669056b144a.png

Résumé#

Le DDD tactique fournit un vocabulaire précis pour traduire les Bounded Contexts stratégiques en code :

  • Les entités portent l’identité et le cycle de vie ; les value objects portent la valeur sans identité propre.

  • Les agrégats garantissent la cohérence transactionnelle et protègent les invariants métier.

  • Les repositories abstraient la persistance, permettant au domaine de rester pur.

  • Les domain events capturent ce qui s’est passé et permettent le découplage entre contextes.

  • Les domain services hébergent la logique inter-entités ; les application services orchestrent les use cases sans logique métier.

  • Les factories encapsulent la création complexe.

La vraie valeur du DDD tactique

Le DDD tactique n’est pas une liste de patterns à appliquer mécaniquement. Sa vraie valeur réside dans l’alignement constant entre le code et le langage du domaine. Chaque invariant de l’agrégat, chaque nom d’événement, chaque méthode de service doit être compréhensible par un expert métier. Quand ce n’est plus le cas, le code a dérivé du domaine — et c’est le moment de refactorer.

Contenu