Modèles d'Architecture

Maîtrisez les modèles d'architecture backend éprouvés, notamment Clean Architecture, Hexagonal Architecture et Domain-Driven Design, pour construire des systèmes maintenables, testables et scalables.

Donné : une limite de service ou un module à architecturer. Produit : une structure en couches avec des règles de dépendances claires, des définitions d'interfaces et des limites de test.

Quand utiliser cette compétence

Concevoir de nouveaux services backend ou microservices à partir de zéro
Refactoriser les applications monolithiques où la logique métier est entrelacée avec les modèles ORM ou les préoccupations HTTP
Établir des contextes limités avant de diviser un système en services
Déboguer les cycles de dépendances où le code d'infrastructure s'infiltre dans la couche domaine
Créer des codebases testables où les tests de cas d'usage ne nécessitent pas une base de données en cours d'exécution
Implémenter les modèles tactiques de domain-driven design (agrégats, objets valeur, événements domaine)

Concepts clés

1. Clean Architecture (Uncle Bob)

Couches (flux de dépendance vers l'intérieur) :

Entities : modèles métier principaux, aucun import de framework
Use Cases : règles métier applicatives, orchestrent les entities
Interface Adapters : contrôleurs, présentateurs, passerelles — traduisent entre les use cases et les formats externes
Frameworks & Drivers : UI, base de données, services externes — tout dans l'anneau le plus externe

Principes clés :

Les dépendances pointent uniquement vers l'intérieur ; les couches intérieures ne connaissent rien des couches extérieures
La logique métier est indépendante des frameworks, des bases de données et des mécanismes de livraison
Chaque limite de couche est franchie via une interface abstraite
Testable sans UI, base de données ou services externes

2. Hexagonal Architecture (Ports et Adapters)

Composants :

Domain Core : la logique métier réside ici, sans framework
Ports : interfaces abstraites qui définissent comment le noyau interagit avec le monde extérieur (driving et driven)
Adapters : implémentations concrètes des ports (adaptateur PostgreSQL, adaptateur Stripe, adaptateur REST)

Avantages :

Échanger les implémentations sans toucher au noyau (par ex., remplacer PostgreSQL par DynamoDB)
Utiliser des adapters en mémoire dans les tests — pas de Docker nécessaire
Décisions technologiques reportées aux bords

3. Domain-Driven Design (DDD)

Modèles stratégiques :

Bounded Contexts : isoler un modèle cohérent pour un sous-domaine ; éviter de partager un seul modèle sur tout le système
Context Mapping : définir comment les contextes se rapportent (Anti-Corruption Layer, Shared Kernel, Open Host Service)
Ubiquitous Language : chaque terme dans le code correspond au terme utilisé par les experts métier

Modèles tactiques :

Entities : objets avec une identité stable qui changent au fil du temps
Value Objects : objets immuables identifiés par leurs attributs (Email, Money, Address)
Aggregates : limites de cohérence ; seule la racine est accessible de l'extérieur
Repositories : persister et reconstituer les agrégats ; abstraire sur le mécanisme de stockage
Domain Events : capturer les choses qui se sont produites dans le domaine ; utilisés pour la coordination entre agrégats

Clean Architecture — Structure des répertoires

app/
├── domain/           # Entities, value objects, interfaces
│   ├── entities/
│   │   ├── user.py
│   │   └── order.py
│   ├── value_objects/
│   │   ├── email.py
│   │   └── money.py
│   └── interfaces/   # Abstract ports (no implementations)
│       ├── user_repository.py
│       └── payment_gateway.py
├── use_cases/        # Application business rules
│   ├── create_user.py
│   ├── process_order.py
│   └── send_notification.py
├── adapters/         # Concrete implementations
│   ├── repositories/
│   │   ├── postgres_user_repository.py
│   │   └── redis_cache_repository.py
│   ├── controllers/
│   │   └── user_controller.py
│   └── gateways/
│       ├── stripe_payment_gateway.py
│       └── sendgrid_email_gateway.py
└── infrastructure/   # Framework wiring, config, DI container
    ├── database.py
    ├── config.py
    └── logging.py

Règle de dépendance en une phrase : chaque instruction import dans domain/ et use_cases/ doit pointer uniquement vers domain/ ; rien dans ces couches ne peut importer depuis adapters/ ou infrastructure/.

Clean Architecture — Implémentation principale

# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    """Core user entity — no framework dependencies."""
    id: str
    email: str
    name: str
    created_at: datetime
    is_active: bool = True

    def deactivate(self):
        self.is_active = False

    def can_place_order(self) -> bool:
        return self.is_active


# domain/interfaces/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
from domain.entities.user import User

class IUserRepository(ABC):
    """Port: defines contract, no implementation details."""

    @abstractmethod
    async def find_by_id(self, user_id: str) -> Optional[User]: ...

    @abstractmethod
    async def find_by_email(self, email: str) -> Optional[User]: ...

    @abstractmethod
    async def save(self, user: User) -> User: ...

    @abstractmethod
    async def delete(self, user_id: str) -> bool: ...


# use_cases/create_user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import uuid
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository

@dataclass
class CreateUserRequest:
    email: str
    name: str

@dataclass
class CreateUserResponse:
    user: Optional[User]
    success: bool
    error: Optional[str] = None

class CreateUserUseCase:
    """Use case: orchestrates business logic, no HTTP or DB details."""

    def __init__(self, user_repository: IUserRepository):
        self.user_repository = user_repository

    async def execute(self, request: CreateUserRequest) -> CreateUserResponse:
        existing = await self.user_repository.find_by_email(request.email)
        if existing:
            return CreateUserResponse(user=None, success=False, error="Email already exists")

        user = User(
            id=str(uuid.uuid4()),
            email=request.email,
            name=request.name,
            created_at=datetime.now(),
        )
        saved_user = await self.user_repository.save(user)
        return CreateUserResponse(user=saved_user, success=True)


# adapters/repositories/postgres_user_repository.py
from domain.interfaces.user_repository import IUserRepository
from domain.entities.user import User
from typing import Optional
import asyncpg

class PostgresUserRepository(IUserRepository):
    """Adapter: PostgreSQL implementation of the user port."""

    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    async def find_by_id(self, user_id: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
            return self._to_entity(row) if row else None

    async def find_by_email(self, email: str) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow("SELECT * FROM users WHERE email = $1", email)
            return self._to_entity(row) if row else None

    async def save(self, user: User) -> User:
        async with self.pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO users (id, email, name, created_at, is_active)
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (id) DO UPDATE
                SET email = $2, name = $3, is_active = $5
                """,
                user.id, user.email, user.name, user.created_at, user.is_active,
            )
        return user

    async def delete(self, user_id: str) -> bool:
        async with self.pool.acquire() as conn:
            result = await conn.execute("DELETE FROM users WHERE id = $1", user_id)
            return result == "DELETE 1"

    def _to_entity(self, row) -> User:
        return User(
            id=row["id"], email=row["email"], name=row["name"],
            created_at=row["created_at"], is_active=row["is_active"],
        )


# adapters/controllers/user_controller.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from use_cases.create_user import CreateUserUseCase, CreateUserRequest

router = APIRouter()

class CreateUserDTO(BaseModel):
    email: str
    name: str

@router.post("/users")
async def create_user(
    dto: CreateUserDTO,
    use_case: CreateUserUseCase = Depends(get_create_user_use_case),
):
    """Controller handles HTTP only — no business logic lives here."""
    response = await use_case.execute(CreateUserRequest(email=dto.email, name=dto.name))
    if not response.success:
        raise HTTPException(status_code=400, detail=response.error)
    return {"user": response.user}

Hexagonal Architecture — Ports et Adapters

# Core domain service — no infrastructure dependencies
class OrderService:
    def __init__(
        self,
        order_repository: OrderRepositoryPort,
        payment_gateway: PaymentGatewayPort,
        notification_service: NotificationPort,
    ):
        self.orders = order_repository
        self.payments = payment_gateway
        self.notifications = notification_service

    async def place_order(self, order: Order) -> OrderResult:
        if not order.is_valid():
            return OrderResult(success=False, error="Invalid order")

        payment = await self.payments.charge(amount=order.total, customer=order.customer_id)
        if not payment.success:
            return OrderResult(success=False, error="Payment failed")

        order.mark_as_paid()
        saved_order = await self.orders.save(order)
        await self.notifications.send(
            to=order.customer_email,
            subject="Order confirmed",
            body=f"Order {order.id} confirmed",
        )
        return OrderResult(success=True, order=saved_order)


# Ports (driving and driven interfaces)
class OrderRepositoryPort(ABC):
    @abstractmethod
    async def save(self, order: Order) -> Order: ...

class PaymentGatewayPort(ABC):
    @abstractmethod
    async def charge(self, amount: Money, customer: str) -> PaymentResult: ...

class NotificationPort(ABC):
    @abstractmethod
    async def send(self, to: str, subject: str, body: str): ...


# Production adapter: Stripe
class StripePaymentAdapter(PaymentGatewayPort):
    def __init__(self, api_key: str):
        import stripe
        stripe.api_key = api_key
        self._stripe = stripe

    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        try:
            charge = self._stripe.Charge.create(
                amount=amount.cents, currency=amount.currency, customer=customer
            )
            return PaymentResult(success=True, transaction_id=charge.id)
        except self._stripe.error.CardError as e:
            return PaymentResult(success=False, error=str(e))


# Test adapter: no external dependencies
class MockPaymentAdapter(PaymentGatewayPort):
    async def charge(self, amount: Money, customer: str) -> PaymentResult:
        return PaymentResult(success=True, transaction_id="mock-txn-123")

DDD — Objets valeur et Agrégats

# Value Objects: immutable, validated at construction
from dataclasses import dataclass

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

    def __post_init__(self):
        if "@" not in self.value or "." not in self.value.split("@")[-1]:
            raise ValueError(f"Invalid email: {self.value}")

@dataclass(frozen=True)
class Money:
    amount: int   # cents
    currency: str

    def __post_init__(self):
        if self.amount < 0:
            raise ValueError("Money amount cannot be negative")
        if self.currency not in {"USD", "EUR", "GBP"}:
            raise ValueError(f"Unsupported currency: {self.currency}")

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Currency mismatch")
        return Money(self.amount + other.amount, self.currency)


# Aggregate root: enforces all invariants for its cluster of entities
class Order:
    def __init__(self, id: str, customer_id: str):
        self.id = id
        self.customer_id = customer_id
        self.items: list[OrderItem] = []
        self.status = OrderStatus.PENDING
        self._events: list[DomainEvent] = []

    def add_item(self, product: Product, quantity: int):
        if self.status != OrderStatus.PENDING:
            raise ValueError("Cannot modify a submitted order")
        item = OrderItem(product=product, quantity=quantity)
        self.items.append(item)
        self._events.append(ItemAddedEvent(order_id=self.id, item=item))

    @property
    def total(self) -> Money:
        totals = [item.subtotal() for item in self.items]
        return sum(totals[1:], totals[0]) if totals else Money(0, "USD")

    def submit(self):
        if not self.items:
            raise ValueError("Cannot submit an empty order")
        if self.status != OrderStatus.PENDING:
            raise ValueError("Order already submitted")
        self.status = OrderStatus.SUBMITTED
        self._events.append(OrderSubmittedEvent(order_id=self.id))

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


# Repository: persist and reconstitute aggregates
class OrderRepository(ABC):
    @abstractmethod
    async def find_by_id(self, order_id: str) -> Optional[Order]: ...

    @abstractmethod
    async def save(self, order: Order) -> None: ...
    # Implementations persist events via pop_events() after writing state

Test — Adapters en mémoire

La marque de la Clean Architecture correctement appliquée est que chaque use case peut être exercé dans un test unitaire simple sans véritable base de données, sans Docker et sans réseau :

# tests/unit/test_create_user.py
import asyncio
from typing import Dict, Optional
from domain.entities.user import User
from domain.interfaces.user_repository import IUserRepository
from use_cases.create_user import CreateUserUseCase, CreateUserRequest


class InMemoryUserRepository(IUserRepository):
    def __init__(self):
        self._store: Dict[str, User] = {}

    async def find_by_id(self, user_id: str) -> Optional[User]:
        return self._store.get(user_id)

    async def find_by_email(self, email: str) -> Optional[User]:
        return next((u for u in self._store.values() if u.email == email), None)

    async def save(self, user: User) -> User:
        self._store[user.id] = user
        return user

    async def delete(self, user_id: str) -> bool:
        return self._store.pop(user_id, None) is not None


async def test_create_user_succeeds():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(user_repository=repo)

    response = await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice"))

    assert response.success
    assert response.user.email == "alice@example.com"
    assert response.user.id is not None


async def test_duplicate_email_rejected():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(user_repository=repo)

    await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice"))
    response = await use_case.execute(CreateUserRequest(email="alice@example.com", name="Alice2"))

    assert not response.success
    assert "already exists" in response.error

Dépannage

Les tests de use case nécessitent une base de données en cours d'exécution

La logique métier s'est échappée dans la couche infrastructure. Déplacez tous les appels de base de données derrière une interface IRepository et injectez une implémentation en mémoire dans les tests (voir la section Test ci-dessus). Le constructeur du use case doit accepter le port abstrait, pas la classe concrète.

Importations circulaires entre les couches

Un symptôme courant est ImportError: cannot import name X entre use_cases et adapters. Cela se produit lorsqu'un use case importe une classe d'adapter concrète au lieu du port abstrait. Appliquez la règle : use_cases/ importe uniquement depuis domain/ (entities et interfaces). Il ne doit jamais importer depuis adapters/ ou infrastructure/.

Les décorateurs de framework apparaissent dans les entities du domaine

Si les annotations SQLAlchemy Column() ou Pydantic Field() apparaissent sur les entities du domaine, l'entity n'est plus pure. Créez un modèle ORM séparé dans adapters/repositories/ et mappez vers/depuis l'entity du domaine dans la méthode _to_entity() du repository.

Toute la logique finit dans les contrôleurs

Lorsque le contrôleur dépasse l'analyse des requêtes et le formatage des réponses, extrayez la logique dans une classe use case. Une méthode de contrôleur ne doit faire que trois choses : parser la requête, appeler un use case, mapper la réponse.

Les objets valeur lèvent des erreurs trop tard

Validez les invariants dans __post_init__ (Python) ou le constructeur afin qu'une Email ou une Money invalide ne puisse pas être construite du tout. Cela expose les données mauvaises à la frontière, pas au cœur de la logique métier.

La contamination du contexte à travers les bounded contexts

Si le contexte Order importe les entities User depuis le contexte Identity, introduisez une Anti-Corruption Layer. Le contexte Order doit détenir son propre objet valeur léger CustomerId et appeler le contexte Identity uniquement via une interface explicite.

Modèles avancés

Pour le mappage des contextes limités DDD détaillé, les arbres de projets multi-services complets, les implémentations Anti-Corruption Layer et les comparaisons avec Onion Architecture, voir :

references/advanced-patterns.md

Compétences connexes

microservices-patterns — Appliquez ces modèles d'architecture lors de la décomposition d'un monolithe en services
cqrs-implementation — Utilisez Clean Architecture comme fondation structurelle pour la séparation commande/requête CQRS
saga-orchestration — Les sagas nécessitent des limites d'agrégat bien définies, que les modèles tactiques DDD fournissent
event-store-design — Les événements domaine produits par les agrégats se versent directement dans une event store

architecture-patterns