11 — Patterns comportementaux appliqués

11 — Patterns comportementaux appliqués#

Les patterns comportementaux s’attaquent à un problème fondamental : comment distribuer les responsabilités entre objets et faire circuler l’information sans créer de couplages rigides ? Ils ne concernent pas la structure statique du code, mais la dynamique de ses interactions à l’exécution.

Strategy — algorithmes interchangeables#

Intention#

Le Strategy remplace les longues chaînes if/elif/else sélectionnant un algorithme par une hiérarchie de classes interchangeables. Le contexte délègue l’exécution à une stratégie injectée, sans savoir laquelle.

Quand l’utiliser#

  • Plusieurs variantes d’un algorithme (tri, compression, pricing, validation)

  • L’algorithme change selon le contexte (profil utilisateur, environnement, configuration)

  • Éliminer un isinstance() ou un switch qui grandit à chaque nouvelle variante

Implémentation Python : PricingStrategy#

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List

@dataclass
class CartItem:
    name: str
    unit_price: float
    quantity: int

class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, items: List[CartItem]) -> float:
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        ...

class StandardPricing(PricingStrategy):
    @property
    def name(self) -> str:
        return "Standard"

    def calculate(self, items: List[CartItem]) -> float:
        return sum(item.unit_price * item.quantity for item in items)

class PremiumPricing(PricingStrategy):
    """Tarif premium : -10% sur le total."""
    @property
    def name(self) -> str:
        return "Premium (-10%)"

    def calculate(self, items: List[CartItem]) -> float:
        total = sum(item.unit_price * item.quantity for item in items)
        return total * 0.90

class BulkDiscountPricing(PricingStrategy):
    """Remise par palier selon la quantité totale."""
    @property
    def name(self) -> str:
        return "Remise volume"

    def calculate(self, items: List[CartItem]) -> float:
        total_qty = sum(item.quantity for item in items)
        total = sum(item.unit_price * item.quantity for item in items)
        if total_qty >= 20:
            return total * 0.75
        elif total_qty >= 10:
            return total * 0.85
        return total

class LoyaltyPricing(PricingStrategy):
    """Fidélité : -5% par année d'ancienneté, plafonné à -30%."""
    def __init__(self, years_member: int):
        self._years = years_member

    @property
    def name(self) -> str:
        return f"Fidélité {self._years} ans"

    def calculate(self, items: List[CartItem]) -> float:
        total = sum(item.unit_price * item.quantity for item in items)
        discount = min(self._years * 0.05, 0.30)
        return total * (1 - discount)


class ShoppingCart:
    def __init__(self, strategy: PricingStrategy):
        self._strategy = strategy
        self._items: List[CartItem] = []

    def set_strategy(self, strategy: PricingStrategy) -> None:
        self._strategy = strategy

    def add_item(self, item: CartItem) -> None:
        self._items.append(item)

    def total(self) -> float:
        return self._strategy.calculate(self._items)

Pièges#

  • Ne pas créer une stratégie pour chaque micro-variation — une fonction lambda suffit parfois.

  • Éviter que les stratégies aient besoin de connaître le contexte (couplage inverse).

Observer — découplage émetteur/récepteur#

Intention#

L’Observer établit une relation un-à-plusieurs : quand un objet change d’état, tous ses abonnés sont notifiés automatiquement. L’émetteur ne connaît pas ses récepteurs. C’est le fondement de l’event-driven programming.

Event Bus maison#

Un Event Bus généralise le pattern Observer : il est central, typé, et permet à n’importe quel module de publier ou s’abonner sans référence directe à l’émetteur.

from collections import defaultdict
from typing import Callable, Dict, List, Any
from dataclasses import dataclass
import time

@dataclass
class Event:
    name: str
    payload: Dict[str, Any]
    timestamp: float = None

    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = time.time()

class EventBus:
    def __init__(self):
        self._handlers: Dict[str, List[Callable]] = defaultdict(list)
        self._history: List[tuple] = []

    def subscribe(self, event_name: str, handler: Callable[[Event], None]) -> None:
        self._handlers[event_name].append(handler)

    def unsubscribe(self, event_name: str, handler: Callable) -> None:
        self._handlers[event_name] = [
            h for h in self._handlers[event_name] if h != handler
        ]

    def publish(self, event: Event) -> None:
        self._history.append((event.timestamp, event.name, event.payload))
        handlers = self._handlers.get(event.name, [])
        for handler in handlers:
            try:
                handler(event)
            except Exception as e:
                print(f"[BUS] Erreur dans {handler.__name__}: {e}")

    @property
    def history(self) -> List[tuple]:
        return list(self._history)

Cas e-commerce : OrderPlaced déclenche 3 handlers indépendants :

bus = EventBus()

def update_inventory(event: Event):
    order_id = event.payload["order_id"]
    items = event.payload["items"]
    print(f"  [STOCK] Réservation stock pour commande {order_id}: {items}")

def send_confirmation_email(event: Event):
    email = event.payload["customer_email"]
    order_id = event.payload["order_id"]
    print(f"  [EMAIL] Confirmation envoyée à {email} pour commande {order_id}")

def track_analytics(event: Event):
    total = event.payload["total"]
    print(f"  [ANALYTICS] Nouvelle commande enregistrée, montant: {total}€")

bus.subscribe("OrderPlaced", update_inventory)
bus.subscribe("OrderPlaced", send_confirmation_email)
bus.subscribe("OrderPlaced", track_analytics)

# L'émetteur ne connaît aucun des handlers
bus.publish(Event("OrderPlaced", {
    "order_id": "ORD-001",
    "customer_email": "alice@example.com",
    "items": ["Livre Python", "Clavier mécanique"],
    "total": 89.90
}))

Pièges#

  • L’ordre d’exécution des handlers n’est pas garanti sans file de priorité.

  • Un handler qui échoue ne doit pas bloquer les autres — toujours try/except.

  • En cas d’abonnements trop nombreux, le débogage devient difficile (« qui a déclenché quoi ? ») — le champ history du bus aide.

Command — encapsulation d’actions#

Intention#

Le Command encapsule une action et ses paramètres dans un objet. Cela permet de : différer l’exécution, construire des historiques undo/redo, composer des macros, et distribuer des tâches dans des queues.

Implémentation : éditeur de texte avec undo/redo#

from abc import ABC, abstractmethod
from typing import List

class Command(ABC):
    @abstractmethod
    def execute(self) -> None:
        ...

    @abstractmethod
    def undo(self) -> None:
        ...

class TextDocument:
    def __init__(self):
        self._content = ""

    @property
    def content(self) -> str:
        return self._content

    def insert(self, position: int, text: str) -> None:
        self._content = self._content[:position] + text + self._content[position:]

    def delete(self, position: int, length: int) -> None:
        self._content = self._content[:position] + self._content[position + length:]


class InsertCommand(Command):
    def __init__(self, doc: TextDocument, position: int, text: str):
        self._doc = doc
        self._position = position
        self._text = text

    def execute(self) -> None:
        self._doc.insert(self._position, self._text)

    def undo(self) -> None:
        self._doc.delete(self._position, len(self._text))


class DeleteCommand(Command):
    def __init__(self, doc: TextDocument, position: int, length: int):
        self._doc = doc
        self._position = position
        self._length = length
        self._deleted_text = ""

    def execute(self) -> None:
        self._deleted_text = self._doc.content[self._position:self._position + self._length]
        self._doc.delete(self._position, self._length)

    def undo(self) -> None:
        self._doc.insert(self._position, self._deleted_text)


class CommandHistory:
    def __init__(self):
        self._history: List[Command] = []
        self._redo_stack: List[Command] = []

    def execute(self, command: Command) -> None:
        command.execute()
        self._history.append(command)
        self._redo_stack.clear()  # Toute nouvelle action vide le redo

    def undo(self) -> bool:
        if not self._history:
            return False
        command = self._history.pop()
        command.undo()
        self._redo_stack.append(command)
        return True

    def redo(self) -> bool:
        if not self._redo_stack:
            return False
        command = self._redo_stack.pop()
        command.execute()
        self._history.append(command)
        return True

Chain of Responsibility — pipeline de traitement#

Intention#

La Chain of Responsibility passe une requête le long d’une chaîne de handlers. Chaque handler décide de traiter la requête et/ou de la passer au suivant. C’est le pattern fondamental des middlewares HTTP.

Implémentation : pipeline de validation et d’authentification#

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, Dict, Any

@dataclass
class HttpRequest:
    method: str
    path: str
    headers: Dict[str, str] = field(default_factory=dict)
    body: Optional[Dict] = None

@dataclass
class HttpResponse:
    status_code: int
    body: Any
    headers: Dict[str, str] = field(default_factory=dict)


class Middleware(ABC):
    def __init__(self):
        self._next: Optional["Middleware"] = None

    def set_next(self, handler: "Middleware") -> "Middleware":
        self._next = handler
        return handler  # Permet le chaînage fluent

    def pass_to_next(self, request: HttpRequest) -> Optional[HttpResponse]:
        if self._next:
            return self._next.handle(request)
        return None

    @abstractmethod
    def handle(self, request: HttpRequest) -> Optional[HttpResponse]:
        ...


class AuthenticationMiddleware(Middleware):
    VALID_TOKENS = {"token_alice", "token_bob", "token_admin"}

    def handle(self, request: HttpRequest) -> Optional[HttpResponse]:
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not token or token not in self.VALID_TOKENS:
            print(f"  [AUTH] Rejeté — token invalide: '{token}'")
            return HttpResponse(401, {"error": "Non authentifié"})
        print(f"  [AUTH] OK — token: {token[:10]}...")
        return self.pass_to_next(request)


class AuthorizationMiddleware(Middleware):
    ADMIN_PATHS = ["/admin/", "/internal/"]

    def handle(self, request: HttpRequest) -> Optional[HttpResponse]:
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        is_admin_path = any(request.path.startswith(p) for p in self.ADMIN_PATHS)
        if is_admin_path and token != "token_admin":
            print(f"  [AUTHZ] Rejeté — accès admin requis pour {request.path}")
            return HttpResponse(403, {"error": "Accès interdit"})
        print(f"  [AUTHZ] OK — accès autorisé à {request.path}")
        return self.pass_to_next(request)


class RateLimitMiddleware(Middleware):
    def __init__(self, max_requests: int = 100):
        super().__init__()
        self._counts: Dict[str, int] = {}
        self._max = max_requests

    def handle(self, request: HttpRequest) -> Optional[HttpResponse]:
        token = request.headers.get("Authorization", "anonymous")
        self._counts[token] = self._counts.get(token, 0) + 1
        if self._counts[token] > self._max:
            print(f"  [RATE] Rejeté — trop de requêtes ({self._counts[token]})")
            return HttpResponse(429, {"error": "Trop de requêtes"})
        print(f"  [RATE] OK — requête #{self._counts[token]}")
        return self.pass_to_next(request)


class RouteHandler(Middleware):
    def handle(self, request: HttpRequest) -> Optional[HttpResponse]:
        print(f"  [ROUTE] Traitement {request.method} {request.path}")
        return HttpResponse(200, {"message": f"OK: {request.path}"})

Construction de la chaîne :

auth = AuthenticationMiddleware()
authz = AuthorizationMiddleware()
rate = RateLimitMiddleware(max_requests=10)
route = RouteHandler()

auth.set_next(authz).set_next(rate).set_next(route)

State — machines à états finis#

Intention#

Le State externalise le comportement dépendant de l’état dans des classes distinctes. Au lieu d’un if state == X: ... elif state == Y: ... qui grandit indéfiniment, chaque état est un objet qui implémente le comportement spécifique à cet état.

Implémentation : commande e-commerce#

from abc import ABC, abstractmethod
from typing import Optional

class OrderContext:  # Forward declaration pour le type
    pass

class OrderState(ABC):
    @abstractmethod
    def confirm(self, order: "OrderContext") -> None:
        ...

    @abstractmethod
    def ship(self, order: "OrderContext") -> None:
        ...

    @abstractmethod
    def deliver(self, order: "OrderContext") -> None:
        ...

    @abstractmethod
    def cancel(self, order: "OrderContext") -> None:
        ...

    @property
    @abstractmethod
    def status(self) -> str:
        ...

    def _reject(self, action: str) -> None:
        print(f"  [STATE] Action '{action}' impossible depuis l'état '{self.status}'")


class PendingState(OrderState):
    @property
    def status(self) -> str:
        return "pending"

    def confirm(self, order: "OrderContext") -> None:
        print("  [STATE] Commande confirmée ✓")
        order.state = ConfirmedState()

    def ship(self, order: "OrderContext") -> None:
        self._reject("ship")

    def deliver(self, order: "OrderContext") -> None:
        self._reject("deliver")

    def cancel(self, order: "OrderContext") -> None:
        print("  [STATE] Commande annulée")
        order.state = CancelledState()


class ConfirmedState(OrderState):
    @property
    def status(self) -> str:
        return "confirmed"

    def confirm(self, order: "OrderContext") -> None:
        self._reject("confirm")

    def ship(self, order: "OrderContext") -> None:
        print("  [STATE] Commande expédiée ✓")
        order.state = ShippedState()

    def deliver(self, order: "OrderContext") -> None:
        self._reject("deliver")

    def cancel(self, order: "OrderContext") -> None:
        print("  [STATE] Commande annulée (remboursement en cours)")
        order.state = CancelledState()


class ShippedState(OrderState):
    @property
    def status(self) -> str:
        return "shipped"

    def confirm(self, order: "OrderContext") -> None:
        self._reject("confirm")

    def ship(self, order: "OrderContext") -> None:
        self._reject("ship")

    def deliver(self, order: "OrderContext") -> None:
        print("  [STATE] Commande livrée ✓")
        order.state = DeliveredState()

    def cancel(self, order: "OrderContext") -> None:
        self._reject("cancel")  # Trop tard pour annuler


class DeliveredState(OrderState):
    @property
    def status(self) -> str:
        return "delivered"

    def confirm(self, order): self._reject("confirm")
    def ship(self, order): self._reject("ship")
    def deliver(self, order): self._reject("deliver")
    def cancel(self, order): self._reject("cancel")


class CancelledState(OrderState):
    @property
    def status(self) -> str:
        return "cancelled"

    def confirm(self, order): self._reject("confirm")
    def ship(self, order): self._reject("ship")
    def deliver(self, order): self._reject("deliver")
    def cancel(self, order): self._reject("cancel")


class OrderContext:
    def __init__(self, order_id: str):
        self.order_id = order_id
        self.state: OrderState = PendingState()

    def confirm(self): self.state.confirm(self)
    def ship(self): self.state.ship(self)
    def deliver(self): self.state.deliver(self)
    def cancel(self): self.state.cancel(self)

    @property
    def status(self) -> str:
        return self.state.status

Template Method — squelette d’algorithme#

Intention#

Le Template Method définit le squelette d’un algorithme dans une méthode de classe de base, en déléguant certaines étapes aux sous-classes. Les étapes invariantes restent dans la classe parente ; les étapes variables sont des méthodes abstraites.

Implémentation : DataExporter#

from abc import ABC, abstractmethod
from typing import List, Dict, Any
import json

class DataExporter(ABC):
    """Template Method : le squelette d'export est fixe."""

    def export(self, data: List[Dict]) -> str:
        """Méthode template — ne pas surcharger."""
        print(f"  [EXPORT] Démarrage export {self.format_name}")
        validated = self._validate(data)
        transformed = self._transform(validated)
        serialized = self._serialize(transformed)
        result = self._add_header(serialized)
        print(f"  [EXPORT] Export {self.format_name} terminé ({len(result)} caractères)")
        return result

    def _validate(self, data: List[Dict]) -> List[Dict]:
        """Étape commune : filtrer les enregistrements nuls."""
        return [row for row in data if row]

    def _transform(self, data: List[Dict]) -> List[Dict]:
        """Hook optionnel — les sous-classes peuvent surcharger."""
        return data

    @abstractmethod
    def _serialize(self, data: List[Dict]) -> str:
        ...

    def _add_header(self, content: str) -> str:
        """Hook optionnel — pas d'en-tête par défaut."""
        return content

    @property
    @abstractmethod
    def format_name(self) -> str:
        ...


class CsvExporter(DataExporter):
    @property
    def format_name(self) -> str:
        return "CSV"

    def _serialize(self, data: List[Dict]) -> str:
        if not data:
            return ""
        headers = list(data[0].keys())
        rows = [",".join(str(row.get(h, "")) for h in headers) for row in data]
        return "\n".join(rows)

    def _add_header(self, content: str) -> str:
        # Pour CSV, on ajoute la ligne d'en-têtes
        return content  # Déjà incluse dans _serialize pour CSV

class JsonExporter(DataExporter):
    @property
    def format_name(self) -> str:
        return "JSON"

    def _serialize(self, data: List[Dict]) -> str:
        return json.dumps(data, ensure_ascii=False, indent=2)

class XmlExporter(DataExporter):
    @property
    def format_name(self) -> str:
        return "XML"

    def _transform(self, data: List[Dict]) -> List[Dict]:
        # Normalise les clés pour XML (pas de tirets, pas d'espaces)
        return [{k.replace(" ", "_").replace("-", "_"): v for k, v in row.items()} for row in data]

    def _serialize(self, data: List[Dict]) -> str:
        lines = ["<records>"]
        for row in data:
            lines.append("  <record>")
            for k, v in row.items():
                lines.append(f"    <{k}>{v}</{k}>")
            lines.append("  </record>")
        lines.append("</records>")
        return "\n".join(lines)

Démonstrations exécutables#

Démo 1 — Strategy : système de pricing avec 4 stratégies#

import matplotlib.pyplot as plt
import seaborn as sns
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List

@dataclass
class CartItem:
    name: str
    unit_price: float
    quantity: int

class PricingStrategy(ABC):
    @abstractmethod
    def calculate(self, items: List[CartItem]) -> float:
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        ...

class StandardPricing(PricingStrategy):
    @property
    def name(self): return "Standard"
    def calculate(self, items):
        return sum(i.unit_price * i.quantity for i in items)

class PremiumPricing(PricingStrategy):
    @property
    def name(self): return "Premium (-10%)"
    def calculate(self, items):
        return sum(i.unit_price * i.quantity for i in items) * 0.90

class BulkDiscountPricing(PricingStrategy):
    @property
    def name(self): return "Volume (-15 à -25%)"
    def calculate(self, items):
        total_qty = sum(i.quantity for i in items)
        total = sum(i.unit_price * i.quantity for i in items)
        if total_qty >= 20: return total * 0.75
        elif total_qty >= 10: return total * 0.85
        return total

class LoyaltyPricing(PricingStrategy):
    def __init__(self, years: int):
        self._years = years
    @property
    def name(self): return f"Fidélité {self._years}ans (-{min(self._years*5, 30)}%)"
    def calculate(self, items):
        total = sum(i.unit_price * i.quantity for i in items)
        return total * (1 - min(self._years * 0.05, 0.30))

# Panier test
cart = [
    CartItem("Livre Python", 35.00, 3),
    CartItem("Clavier mécanique", 120.00, 1),
    CartItem("Souris ergonomique", 75.00, 2),
]

strategies = [
    StandardPricing(),
    PremiumPricing(),
    BulkDiscountPricing(),
    LoyaltyPricing(4),
]

totals = []
labels = []
standard_total = None

for strategy in strategies:
    total = strategy.calculate(cart)
    totals.append(total)
    labels.append(strategy.name)
    if standard_total is None:
        standard_total = total
    savings = standard_total - total
    print(f"{strategy.name:30s} → {total:8.2f}€  (économie: {savings:.2f}€)")

# Visualisation
sns.set_theme(style="whitegrid", palette="muted", font_scale=1.1)
fig, ax = plt.subplots(figsize=(10, 5))

colors = sns.color_palette("muted", len(totals))
bars = ax.barh(labels, totals, color=colors, height=0.55, edgecolor="white")

for bar, total in zip(bars, totals):
    ax.text(bar.get_width() + 3, bar.get_y() + bar.get_height() / 2,
            f"{total:.2f}€", va="center", fontsize=10, fontweight="bold")

ax.axvline(standard_total, color="#C44E52", linestyle="--", linewidth=1.5,
           label=f"Prix standard ({standard_total:.2f}€)")
ax.set_xlabel("Total panier (€)")
ax.set_title("Comparaison des stratégies de pricing\n(même panier, 4 stratégies)", fontsize=13, fontweight="bold")
ax.legend(fontsize=9)
ax.set_xlim(0, max(totals) * 1.18)
plt.savefig("pricing_strategies.png", dpi=120, bbox_inches="tight")
plt.show()

Standard                       →   375.00€  (économie: 0.00€)
Premium (-10%)                 →   337.50€  (économie: 37.50€)
Volume (-15 à -25%)            →   375.00€  (économie: 0.00€)
Fidélité 4ans (-20%)           →   300.00€  (économie: 75.00€)
_images/b5b52aeac30db35778dacfb133bdd211c15c303b60dfeb5ad941ea379143d309.png

Démo 2 — Observer/EventBus : bus d’événements e-commerce#

from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Any
import time

@dataclass
class Event:
    name: str
    payload: Dict[str, Any]
    timestamp: float = field(default_factory=time.time)

class EventBus:
    def __init__(self):
        self._handlers: Dict[str, List[Callable]] = defaultdict(list)
        self._log: List[str] = []

    def subscribe(self, event_name: str, handler: Callable) -> None:
        self._handlers[event_name].append(handler)

    def publish(self, event: Event) -> None:
        self._log.append(f"[BUS] Publication: {event.name}")
        for handler in self._handlers.get(event.name, []):
            try:
                handler(event)
            except Exception as e:
                self._log.append(f"[BUS] Erreur dans {handler.__name__}: {e}")

    def print_log(self):
        for entry in self._log:
            print(entry)


bus = EventBus()

# Handler 1 : mise à jour du stock
def update_inventory(event: Event):
    items = event.payload.get("items", [])
    for item in items:
        print(f"  [STOCK] Réduction stock: {item['name']} x{item['qty']}")

# Handler 2 : envoi d'email
def send_confirmation(event: Event):
    email = event.payload.get("customer_email")
    order_id = event.payload.get("order_id")
    print(f"  [EMAIL] Confirmation envoyée à {email} — commande {order_id}")

# Handler 3 : analytics
def record_analytics(event: Event):
    total = event.payload.get("total", 0)
    print(f"  [ANALYTICS] Événement '{event.name}' — valeur: {total:.2f}€")

# Handler 4 : réassort automatique
def check_restock(event: Event):
    items = event.payload.get("items", [])
    for item in items:
        if item["qty"] > 5:
            print(f"  [RESTOCK] Alerte réassort déclenchée pour: {item['name']}")

bus.subscribe("OrderPlaced", update_inventory)
bus.subscribe("OrderPlaced", send_confirmation)
bus.subscribe("OrderPlaced", record_analytics)
bus.subscribe("OrderPlaced", check_restock)
bus.subscribe("OrderCancelled", record_analytics)

print("=== Événement 1 : nouvelle commande ===")
bus.publish(Event("OrderPlaced", {
    "order_id": "ORD-2024-001",
    "customer_email": "alice@example.com",
    "items": [
        {"name": "Livre Architecture Logicielle", "qty": 2, "price": 35.0},
        {"name": "Clavier mécanique", "qty": 8, "price": 120.0},
    ],
    "total": 1030.0,
}))

print("\n=== Événement 2 : commande annulée ===")
bus.publish(Event("OrderCancelled", {
    "order_id": "ORD-2024-002",
    "total": 89.90,
}))

print(f"\n=== {len(bus._log)} événements traités ===")

=== Événement 1 : nouvelle commande ===
  [STOCK] Réduction stock: Livre Architecture Logicielle x2
  [STOCK] Réduction stock: Clavier mécanique x8
  [EMAIL] Confirmation envoyée à alice@example.com — commande ORD-2024-001
  [ANALYTICS] Événement 'OrderPlaced' — valeur: 1030.00€
  [RESTOCK] Alerte réassort déclenchée pour: Clavier mécanique

=== Événement 2 : commande annulée ===
  [ANALYTICS] Événement 'OrderCancelled' — valeur: 89.90€

=== 2 événements traités ===

Démo 3 — Chain of Responsibility : pipeline de validation HTTP#

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, Dict, Any

@dataclass
class HttpRequest:
    method: str
    path: str
    headers: Dict[str, str] = field(default_factory=dict)
    body: Optional[Dict] = None

@dataclass
class HttpResponse:
    status_code: int
    body: Any

class Middleware(ABC):
    def __init__(self):
        self._next: Optional["Middleware"] = None

    def set_next(self, handler: "Middleware") -> "Middleware":
        self._next = handler
        return handler

    def pass_to_next(self, request: HttpRequest) -> HttpResponse:
        if self._next:
            return self._next.handle(request)
        return HttpResponse(500, {"error": "Aucun handler final"})

    @abstractmethod
    def handle(self, request: HttpRequest) -> HttpResponse:
        ...


class AuthMiddleware(Middleware):
    TOKENS = {"tok_alice": "alice", "tok_admin": "admin"}

    def handle(self, request: HttpRequest) -> HttpResponse:
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if token not in self.TOKENS:
            print(f"  [AUTH] 401 — token inconnu")
            return HttpResponse(401, {"error": "Non authentifié"})
        request.headers["_user"] = self.TOKENS[token]
        print(f"  [AUTH] 200 — utilisateur: {self.TOKENS[token]}")
        return self.pass_to_next(request)


class AuthzMiddleware(Middleware):
    ADMIN_PATHS = ["/admin"]

    def handle(self, request: HttpRequest) -> HttpResponse:
        user = request.headers.get("_user")
        needs_admin = any(request.path.startswith(p) for p in self.ADMIN_PATHS)
        if needs_admin and user != "admin":
            print(f"  [AUTHZ] 403 — accès admin requis")
            return HttpResponse(403, {"error": "Interdit"})
        print(f"  [AUTHZ] OK — {user} accède à {request.path}")
        return self.pass_to_next(request)


class LoggingMiddleware(Middleware):
    def handle(self, request: HttpRequest) -> HttpResponse:
        print(f"  [LOG] {request.method} {request.path}")
        response = self.pass_to_next(request)
        print(f"  [LOG] Réponse: {response.status_code}")
        return response


class RouteHandler(Middleware):
    def handle(self, request: HttpRequest) -> HttpResponse:
        print(f"  [ROUTE] Traitement de la requête")
        return HttpResponse(200, {"data": f"Contenu de {request.path}"})


# Construction de la chaîne
log_mw = LoggingMiddleware()
auth_mw = AuthMiddleware()
authz_mw = AuthzMiddleware()
route = RouteHandler()
log_mw.set_next(auth_mw).set_next(authz_mw).set_next(route)

scenarios = [
    ("Accès utilisateur normal", HttpRequest("GET", "/api/users", {"Authorization": "Bearer tok_alice"})),
    ("Token invalide",           HttpRequest("GET", "/api/users", {"Authorization": "Bearer bad_token"})),
    ("Accès admin refusé",       HttpRequest("GET", "/admin/stats", {"Authorization": "Bearer tok_alice"})),
    ("Accès admin autorisé",     HttpRequest("GET", "/admin/stats", {"Authorization": "Bearer tok_admin"})),
]

for title, req in scenarios:
    print(f"\n>>> Scénario : {title}")
    resp = log_mw.handle(req)
    print(f"    Résultat final : HTTP {resp.status_code}")

>>> Scénario : Accès utilisateur normal
  [LOG] GET /api/users
  [AUTH] 200 — utilisateur: alice
  [AUTHZ] OK — alice accède à /api/users
  [ROUTE] Traitement de la requête
  [LOG] Réponse: 200
    Résultat final : HTTP 200

>>> Scénario : Token invalide
  [LOG] GET /api/users
  [AUTH] 401 — token inconnu
  [LOG] Réponse: 401
    Résultat final : HTTP 401

>>> Scénario : Accès admin refusé
  [LOG] GET /admin/stats
  [AUTH] 200 — utilisateur: alice
  [AUTHZ] 403 — accès admin requis
  [LOG] Réponse: 403
    Résultat final : HTTP 403

>>> Scénario : Accès admin autorisé
  [LOG] GET /admin/stats
  [AUTH] 200 — utilisateur: admin
  [AUTHZ] OK — admin accède à /admin/stats
  [ROUTE] Traitement de la requête
  [LOG] Réponse: 200
    Résultat final : HTTP 200

Démo 4 — Radar des patterns comportementaux#

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

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

patterns = ["Strategy", "Observer", "Command", "Chain of\nResponsibility", "State", "Template\nMethod"]
metrics = ["Couplage faible", "Complexité implémentation", "Flexibilité", "Testabilité", "Courbe d'apprentissage"]

# Scores sur 10 pour chaque (pattern, métrique)
scores = np.array([
    [9, 3, 9, 9, 2],   # Strategy
    [9, 5, 8, 7, 4],   # Observer
    [8, 6, 9, 8, 5],   # Command
    [7, 4, 8, 8, 3],   # Chain of Resp
    [7, 7, 7, 8, 6],   # State
    [6, 3, 6, 8, 2],   # Template Method
])

n_metrics = len(metrics)
angles = np.linspace(0, 2 * np.pi, n_metrics, endpoint=False).tolist()
angles += angles[:1]

colors_palette = sns.color_palette("muted", len(patterns))

fig, ax = plt.subplots(figsize=(10, 8), subplot_kw=dict(polar=True))
ax.set_facecolor("#F8F9FA")

for i, (pattern, score) in enumerate(zip(patterns, scores)):
    values = score.tolist()
    values += values[:1]
    ax.plot(angles, values, "o-", linewidth=2, color=colors_palette[i],
            label=pattern.replace("\n", " "), alpha=0.9)
    ax.fill(angles, values, alpha=0.08, color=colors_palette[i])

ax.set_xticks(angles[:-1])
ax.set_xticklabels(metrics, size=10, fontweight="bold")
ax.set_ylim(0, 10)
ax.set_yticks([2, 4, 6, 8, 10])
ax.set_yticklabels(["2", "4", "6", "8", "10"], size=8)
ax.grid(True, alpha=0.4)

ax.legend(loc="upper right", bbox_to_anchor=(1.35, 1.15),
          fontsize=9, framealpha=0.9)
ax.set_title("Profil des patterns comportementaux\n(couplage, complexité, flexibilité, testabilité)",
             fontsize=12, fontweight="bold", pad=20, color="#2C3E50")
plt.savefig("patterns_radar.png", dpi=120, bbox_inches="tight")
plt.show()
_images/ced91a6c93071772057ff793843f390189d1ec87c2afb96c1071774db7843c71.png

Résumé#

Les patterns comportementaux partagent un objectif commun : distribuer les responsabilités sans créer de dépendances directes entre les objets. Chacun attaque une source spécifique de rigidité :

  • Strategy — élimine les if/elif algorithmiques. Le changement d’algorithme devient un simple changement d’objet injecté.

  • Observer — découple émetteur et récepteurs. Un événement peut avoir zéro ou cent handlers sans que l’émetteur en sache rien.

  • Command — objectifie les actions. Une fois une action encapsulée, on peut la différer, la rejouer, l’annuler, la logger.

  • Chain of Responsibility — structure les pipelines de traitement. Chaque maillon est indépendant et testable séparément.

  • State — remplace les flags booléens et les if state == ... enchevêtrés par des objets état cohérents.

  • Template Method — fixe la structure algorithmique tout en laissant les détails aux sous-classes. C’est la réutilisation de code par héritage la plus légitime.

À retenir

Strategy et State se ressemblent structurellement (tous deux utilisent la composition et l’injection d’objets), mais leur intention diffère radicalement. Strategy choisit comment faire quelque chose (algorithme interchangeable) ; State détermine ce que fait un objet selon sa situation courante (comportement dépendant du contexte).

Piège courant

L’Observer peut créer des « chaînes d’événements » difficiles à tracer : un événement déclenche des handlers qui publient d’autres événements, qui déclenchent d’autres handlers… Le débogage devient un cauchemar. Toujours journaliser les publications dans le bus et limiter les handlers qui publient à leur tour des événements.

Bonne pratique

Le Template Method est souvent la forme la plus simple de réutilisation : une classe de base définit le flux, les sous-classes remplissent les blancs. Il faut cependant éviter les hiérarchies profondes — si une sous-classe surcharge plus de la moitié des méthodes template, la classe de base est probablement trop rigide et Strategy serait un meilleur choix.

Contenu