Cours 10 : Architecture Logicielle en Python

1. Clean Architecture

1.1 Principes fondamentaux

La Clean Architecture (Robert C. Martin) organise le code en cercles concentriques :

┌─────────────────────────────────────┐
│         Frameworks & Drivers        │
│  ┌───────────────────────────────┐  │
│  │      Interface Adapters       │  │
│  │  ┌─────────────────────────┐  │  │
│  │  │    Application Layer    │  │  │
│  │  │  ┌───────────────────┐  │  │  │
│  │  │  │   Domain Layer    │  │  │  │
│  │  │  │  (Entities)       │  │  │  │
│  │  │  └───────────────────┘  │  │  │
│  │  └─────────────────────────┘  │  │
│  └───────────────────────────────┘  │
└─────────────────────────────────────┘

Règle d'or : Les dépendances vont de l'exterieur vers l'interieur. Le code métier (domaine) ne dépend de rien d'autre que du langage lui-même.

1.2 Entities (Domaine)

from dataclasses import dataclass
from uuid import UUID, uuid4
from abc import ABC, abstractmethod

@dataclass(frozen=True)
class Email:
    address: str
    def __post_init__(self):
        if "@" not in self.address:
            raise ValueError(f"Invalid email: {self.address}")

class User:
    def __init__(self, id: UUID, email: Email, name: str):
        self.id = id
        self.email = email
        self.name = name

    def change_email(self, email: str) -> None:
        self.email = Email(email)

1.3 Use Cases (Application)

class CreateUserUseCase:
    def __init__(self, repo: "UserRepository"):
        self.repo = repo

    def execute(self, name: str, email: str) -> User:
        user = User(id=uuid4(), email=Email(email), name=name)
        self.repo.save(user)
        return user

class UserRepository(ABC):
    @abstractmethod
    def save(self, user: User) -> None: ...
    @abstractmethod
    def find_by_id(self, id: UUID) -> User: ...

1.4 Infrastructure (Gateways)

from sqlalchemy.orm import Session

class PostgresUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session

    def save(self, user: User) -> None:
        model = UserModel(id=str(user.id), name=user.name, email=user.email.address)
        self.session.add(model)
        self.session.commit()

    def find_by_id(self, id: UUID) -> User:
        model = self.session.query(UserModel).filter_by(id=str(id)).first()
        return User(id=UUID(model.id), email=Email(model.email), name=model.name)

1.5 Interfaces (API)

from fastapi import FastAPI, Depends

app = FastAPI()

def get_user_repo() -> UserRepository:
    return PostgresUserRepository(get_session())

@app.post("/users")
def create_user(name: str, email: str, repo: UserRepository = Depends(get_user_repo)):
    use_case = CreateUserUseCase(repo)
    return use_case.execute(name, email)

2. Domain-Driven Design (DDD)

2.1 Aggregates

Un Aggregate est un cluster d'objets du domaine traité comme une unité cohérente :

from abc import ABC
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from enum import Enum

class OrderStatus(Enum):
    PENDING = "pending"
    PAID = "paid"
    SHIPPED = "shipped"
    CANCELLED = "cancelled"

@dataclass(frozen=True)
class OrderLine:
    product_id: UUID
    quantity: int
    price: float

class Order:
    def __init__(self, customer_id: UUID):
        self.id = uuid4()
        self.customer_id = customer_id
        self.status = OrderStatus.PENDING
        self.items: list[OrderLine] = []
        self._events: list[DomainEvent] = []

    def add_item(self, product_id: UUID, quantity: int, price: float) -> None:
        self.items.append(OrderLine(product_id, quantity, price))
        self._events.append(ItemAddedToOrder(self.id, product_id, quantity))

    def pay(self) -> None:
        if self.status != OrderStatus.PENDING:
            raise DomainException("Order already processed")
        self.status = OrderStatus.PAID
        self._events.append(OrderPaid(self.id))

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

2.2 Domain Events

from dataclasses import dataclass
from datetime import datetime

class DomainEvent(ABC):
    occurred_at: datetime

@dataclass
class ItemAddedToOrder(DomainEvent):
    order_id: UUID
    product_id: UUID
    quantity: int
    occurred_at: datetime = field(default_factory=datetime.utcnow)

@dataclass
class OrderPaid(DomainEvent):
    order_id: UUID
    occurred_at: datetime = field(default_factory=datetime.utcnow)

class DomainException(Exception):
    pass

2.3 Repositories

class OrderRepository(ABC):
    @abstractmethod
    def save(self, order: Order) -> None: ...
    @abstractmethod
    def find_by_id(self, id: UUID) -> Order: ...
    @abstractmethod
    def delete(self, id: UUID) -> None: ...

2.4 Value Objects

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

    def __post_init__(self):
        if len(self.zip_code) != 5 or not self.zip_code.isdigit():
            raise ValueError(f"Invalid zip code: {self.zip_code}")

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

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

3. Hexagonal Architecture (Ports & Adapters)

3.1 Ports (Interfaces)

class PaymentGateway(ABC):
    @abstractmethod
    def charge(self, amount: Money, token: str) -> str: ...
    @abstractmethod
    def refund(self, transaction_id: str) -> bool: ...

class EmailService(ABC):
    @abstractmethod
    def send(self, to: str, subject: str, body: str) -> None: ...

3.2 Adapters (Implémentations)

import stripe

class StripePaymentGateway(PaymentGateway):
    def __init__(self, api_key: str):
        stripe.api_key = api_key

    def charge(self, amount: Money, token: str) -> str:
        charge = stripe.Charge.create(
            amount=int(amount.amount * 100),
            currency=amount.currency.lower(),
            source=token,
        )
        return charge.id

    def refund(self, transaction_id: str) -> bool:
        refund = stripe.Refund.create(charge=transaction_id)
        return refund.status == "succeeded"

class SendGridEmailService(EmailService):
    def __init__(self, api_key: str):
        self.client = sendgrid.SendGridAPIClient(api_key)

    def send(self, to: str, subject: str, body: str) -> None:
        message = Mail(from_email="noreply@example.com", to_emails=to, subject=subject, html_content=body)
        self.client.send(message)

3.3 Wiring avec DI

from dependency_injector import containers, providers

class ApplicationContainer(containers.DeclarativeContainer):
    config = providers.Configuration()

    # Adapters
    payment_gateway = providers.Singleton(
        StripePaymentGateway,
        api_key=config.stripe.api_key,
    )
    email_service = providers.Singleton(
        SendGridEmailService,
        api_key=config.sendgrid.api_key,
    )

    # Repositories
    order_repo = providers.Singleton(PostgresOrderRepository, session=...)

    # Use Cases
    create_order_uc = providers.Factory(
        CreateOrderUseCase,
        order_repo=order_repo,
        payment_gateway=payment_gateway,
    )

4. Dependency Injection

4.1 FastAPI DI

from fastapi import FastAPI, Depends

app = FastAPI()

# Dépendance
def get_db() -> Generator:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Service
class UserService:
    def __init__(self, db: Session):
        self.db = db

def get_user_service(db: Session = Depends(get_db)) -> UserService:
    return UserService(db)

# Route
@app.get("/users/{id}")
def get_user(id: int, service: UserService = Depends(get_user_service)):
    return service.get_user(id)

4.2 dependency-injector

from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()
    db = providers.Singleton(Database, url=config.db.url)
    user_repo = providers.Factory(UserRepository, db=db)
    user_service = providers.Factory(UserService, repo=user_repo)

@inject
def create_user(name: str, email: str, service: UserService = Provide[Container.user_service]):
    return service.create_user(name, email)

container = Container()
container.config.db.url.from_env("DATABASE_URL")
container.wire(modules=[__name__])

5. Project Structure

project/
├── src/
│   ├── domain/
│   │   ├── __init__.py
│   │   ├── entities.py
│   │   ├── value_objects.py
│   │   └── events.py
│   ├── application/
│   │   ├── __init__.py
│   │   ├── ports/
│   │   │   ├── __init__.py
│   │   │   └── repositories.py
│   │   └── use_cases/
│   │       ├── __init__.py
│   │       └── create_user.py
│   ├── infrastructure/
│   │   ├── __init__.py
│   │   ├── persistence/
│   │   ├── messaging/
│   │   └── external/
│   └── interfaces/
│       ├── __init__.py
│       ├── api/
│       └── cli/
├── tests/
├── docker-compose.yml
└── pyproject.toml

6. Microservices Patterns

Diagramme en cours de génération...

7. Event-Driven avec FastStream

from faststream import FastStream
from faststream.kafka import KafkaBroker

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

@publisher("order-confirmed")
@broker.subscriber("order-created")
async def on_order_created(msg: OrderCreated):
    process_order(msg)
    return OrderConfirmed(order_id=msg.order_id)

@broker.subscriber("payment-processed")
async def on_payment_processed(msg: PaymentProcessed):
    await send_notification(msg.user_id, f"Payment of {msg.amount} confirmed")

8. Tests d'architecture

import pytest
from pytest_arch import rule

def test_domain_does_not_import_infrastructure():
    rule("domain").should_not_depend_on("infrastructure").check("src/")

def test_use_cases_only_depend_on_domain():
    rule("application").should_only_depend_on("domain", "typing").check("src/")

def test_interfaces_can_depend_on_application():
    rule("interfaces").may_depend_on("application", "infrastructure").check("src/")

9. Diagramme : Architecture globale

Diagramme en cours de génération...

10. Bonnes pratiques

1. Toujours commencer par le domaine : Les entités et use cases avant toute considération technique
2. Interfaces explicites : Tous les ports sont des ABC ou Protocols
3. DI partout : Pas de new() dans les use cases
4. Tests par couche : Tester chaque couche isolément avec des mocks
5. Events pour le couplage faible : Communication inter-service asynchrone
6. Bounded Contexts : Délimiter clairement les contextes DDD