Les motifs de conception Python pour une architecture propre
Construisez des applications Python maintenables avec les modèles de conception SOLID
Architecture propre a révolutionné la manière dont les développeurs construisent des applications évolutives et maintenables en mettant l’accent sur la séparation des responsabilités et la gestion des dépendances.
En Python, ces principes s’associent à la nature dynamique du langage pour créer des systèmes flexibles et testables qui évoluent avec les besoins métier sans devenir une dette technique.

Comprendre l’architecture propre en Python
L’architecture propre, introduite par Robert C. Martin (Uncle Bob), organise le logiciel en couches concentriques où les dépendances pointent vers l’intérieur vers la logique métier centrale. Ce modèle architectural garantit que les règles métier critiques de votre application restent indépendantes des frameworks, des bases de données et des services externes.
La philosophie centrale
Le principe fondamental est simple mais puissant : la logique métier ne doit pas dépendre de l’infrastructure. Vos entités métier, cas d’utilisation et règles métier doivent fonctionner indépendamment de savoir si vous utilisez PostgreSQL ou MongoDB, FastAPI ou Flask, AWS ou Azure.
En Python, cette philosophie s’aligne parfaitement avec la “duck typing” et la programmation orientée protocole du langage, permettant une séparation claire sans la cérémonie requise dans les langages typés statiquement.
Les quatre couches de l’architecture propre
Couche Entités (Domaine) : Objets métier purs avec des règles métier applicables à l’ensemble de l’entreprise. Ce sont des POJO (Plain Old Python Objects) sans dépendances externes.
Couche Cas d’utilisation (Application) : Règles métier spécifiques à l’application orchestrant le flux de données entre les entités et les services externes.
Couche Adapteurs d’interface : Convertit les données entre le format le plus pratique pour les cas d’utilisation et les entités, et le format requis par les agences externes.
Couche Cadres et pilotes : Tous les détails externes tels que les bases de données, les cadres web et les API externes.
Principes SOLID en Python
Les principes SOLID forment la base de l’architecture propre. Explorons comment chacun de ces principes s’incarne en Python. Pour un aperçu complet des modèles de conception en Python, consultez le Guide des modèles de conception Python.
Principe de responsabilité unique (SRP)
Chaque classe devrait avoir une seule raison de changer :
# Mauvais : Plusieurs responsabilités
class UserManager:
def create_user(self, user_data):
# Créer un utilisateur
pass
def send_welcome_email(self, user):
# Envoyer un email
pass
def log_creation(self, user):
# Enregistrer dans un fichier
pass
# Bon : Responsabilités séparées
class UserService:
def __init__(self, repository, email_service, logger):
self.repository = repository
self.email_service = email_service
self.logger = logger
def create_user(self, user_data):
user = User(**user_data)
self.repository.save(user)
self.email_service.send_welcome(user)
self.logger.info(f"Utilisateur créé : {user.id}")
return user
Principe ouvert/fermé (OCP)
Les entités logicielles devraient être ouvertes à l’extension mais fermées à la modification :
from abc import ABC, abstractmethod
from typing import Protocol
# Utilisation du Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
def process_payment(self, amount: float) -> bool:
...
class CreditCardProcessor:
def process_payment(self, amount: float) -> bool:
# Logique de carte de crédit
return True
class PayPalProcessor:
def process_payment(self, amount: float) -> bool:
# Logique PayPal
return True
# Extensible sans modifier le code existant
class CryptoProcessor:
def process_payment(self, amount: float) -> bool:
# Logique de cryptomonnaie
return True
Principe de substitution de Liskov (LSP)
Les objets devraient être substituables par leurs sous-types sans que le programme ne soit rompu :
from abc import ABC, abstractmethod
class DataStore(ABC):
@abstractmethod
def save(self, key: str, value: str) -> None:
pass
@abstractmethod
def get(self, key: str) -> str:
pass
class PostgreSQLStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implémentation PostgreSQL
pass
def get(self, key: str) -> str:
# Implémentation PostgreSQL
return ""
class RedisStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implémentation Redis
pass
def get(self, key: str) -> str:
# Implémentation Redis
return ""
# Les deux peuvent être utilisés interchangeables
def process_data(store: DataStore, key: str, value: str):
store.save(key, value)
return store.get(key)
Principe de ségrégation des interfaces (ISP)
Les clients ne devraient pas être contraints de dépendre d’interfaces qu’ils n’utilisent pas :
# Mauvais : Interface trop lourde
class Worker(ABC):
@abstractmethod
def work(self): pass
@abstractmethod
def eat(self): pass
@abstractmethod
def sleep(self): pass
# Bon : Interfaces ségrégées
class Workable(Protocol):
def work(self) -> None: ...
class Eatable(Protocol):
def eat(self) -> None: ...
class Human:
def work(self) -> None:
print("Travaillant")
def eat(self) -> None:
print("Mangeant")
class Robot:
def work(self) -> None:
print("Travaillant")
# Aucune méthode eat nécessaire
Principe d’inversion des dépendances (DIP)
Les modules de haut niveau ne devraient pas dépendre des modules de bas niveau. Les deux devraient dépendre d’abstractions :
from typing import Protocol
# Abstraction
class EmailSender(Protocol):
def send(self, to: str, subject: str, body: str) -> None:
...
# Module de bas niveau
class SMTPEmailSender:
def send(self, to: str, subject: str, body: str) -> None:
# Implémentation SMTP
pass
# Module de haut niveau dépendant de l'abstraction
class UserRegistrationService:
def __init__(self, email_sender: EmailSender):
self.email_sender = email_sender
def register(self, email: str, name: str):
# Logique d'enregistrement
self.email_sender.send(
to=email,
subject="Bienvenue !",
body=f"Bonjour {name}"
)
Modèle de Répository : Abstraction de l’accès aux données
Le Modèle de Répository fournit une interface similaire à une collection pour accéder aux objets métier, cachant les détails du stockage des données.
Implémentation de base du Répository
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4
@dataclass
class User:
id: UUID
email: str
name: str
is_active: bool = True
class UserRepository(ABC):
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
@abstractmethod
def get_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def list_all(self) -> List[User]:
pass
@abstractmethod
def delete(self, user_id: UUID) -> bool:
pass
Implémentation avec SQLAlchemy
from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(PGUUID(as_uuid=True), primary_key=True)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
class SQLAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
user_model = UserModel(
id=user.id,
email=user.email,
name=user.name,
is_active=user.is_active
)
self.session.add(user_model)
self.session.commit()
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.id == user_id
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def get_by_email(self, email: str) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.email == email
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def list_all(self) -> List[User]:
users = self.session.query(UserModel).all()
return [
User(
id=u.id,
email=u.email,
name=u.name,
is_active=u.is_active
)
for u in users
]
def delete(self, user_id: UUID) -> bool:
result = self.session.query(UserModel).filter(
UserModel.id == user_id
).delete()
self.session.commit()
return result > 0
Répository en mémoire pour les tests
class InMemoryUserRepository(UserRepository):
def __init__(self):
self.users: dict[UUID, User] = {}
def save(self, user: User) -> User:
self.users[user.id] = user
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
return self.users.get(user_id)
def get_by_email(self, email: str) -> Optional[User]:
for user in self.users.values():
if user.email == email:
return user
return None
def list_all(self) -> List[User]:
return list(self.users.values())
def delete(self, user_id: UUID) -> bool:
if user_id in self.users:
del self.users[user_id]
return True
return False
Couche Service : Orchestration de la logique métier
La Couche Service implémente les cas d’utilisation et orchestre le flux entre les répositories, les services externes et la logique métier.
from typing import Optional
from uuid import uuid4
class UserAlreadyExistsError(Exception):
pass
class UserNotFoundError(Exception):
pass
class UserService:
def __init__(
self,
user_repository: UserRepository,
email_service: EmailSender,
event_publisher: 'EventPublisher'
):
self.user_repository = user_repository
self.email_service = email_service
self.event_publisher = event_publisher
def register_user(self, email: str, name: str) -> User:
# Vérifier si l'utilisateur existe
existing_user = self.user_repository.get_by_email(email)
if existing_user:
raise UserAlreadyExistsError(f"L'utilisateur avec l'email {email} existe déjà")
# Créer un nouvel utilisateur
user = User(
id=uuid4(),
email=email,
name=name,
is_active=True
)
# Enregistrer dans le répository
user = self.user_repository.save(user)
# Envoyer un email de bienvenue
self.email_service.send(
to=user.email,
subject="Bienvenue !",
body=f"Bonjour {user.name}, bienvenue sur notre plateforme !"
)
# Publier un événement
self.event_publisher.publish('user.registered', {
'user_id': str(user.id),
'email': user.email
})
return user
def deactivate_user(self, user_id: UUID) -> User:
user = self.user_repository.get_by_id(user_id)
if not user:
raise UserNotFoundError(f"L'utilisateur {user_id} non trouvé")
user.is_active = False
user = self.user_repository.save(user)
self.event_publisher.publish('user.deactivated', {
'user_id': str(user.id)
})
return user
Injection de dépendances en Python
La nature dynamique de Python rend l’injection de dépendances simple sans nécessiter de cadres lourds.
Injection par constructeur
class OrderService:
def __init__(
self,
order_repository: 'OrderRepository',
payment_processor: PaymentProcessor,
notification_service: 'NotificationService'
):
self.order_repository = order_repository
self.payment_processor = payment_processor
self.notification_service = notification_service
def place_order(self, order_data: dict):
# Utiliser les dépendances injectées
pass
Conteneur de dépendances simple
from typing import Dict, Type, Callable, Any
class Container:
def __init__(self):
self._services: Dict[Type, Callable] = {}
self._singletons: Dict[Type, Any] = {}
def register(self, interface: Type, factory: Callable):
self._services[interface] = factory
def register_singleton(self, interface: Type, instance: Any):
self._singletons[interface] = instance
def resolve(self, interface: Type):
if interface in self._singletons:
return self._singletons[interface]
factory = self._services.get(interface)
if factory:
return factory(self)
raise ValueError(f"Aucune inscription trouvée pour {interface}")
# Utilisation
def create_container() -> Container:
container = Container()
# Enregistrer les services
container.register_singleton(
Session,
sessionmaker(bind=create_engine('postgresql://...'))()
)
container.register(
UserRepository,
lambda c: SQLAlchemyUserRepository(c.resolve(Session))
)
container.register(
EmailSender,
lambda c: SMTPEmailSender()
)
container.register(
UserService,
lambda c: UserService(
c.resolve(UserRepository),
c.resolve(EmailSender),
c.resolve(EventPublisher)
)
)
return container
Architecture Hexagonale (Ports et Adapteurs)
L’Architecture Hexagonale place la logique métier au centre avec des adaptateurs gérant la communication externe.
Définir les Ports (Interfaces)
# Port d'entrée (Principal)
class CreateUserUseCase(Protocol):
def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
...
# Port de sortie (Secondaire)
class UserPersistencePort(Protocol):
def save(self, user: User) -> User:
...
def find_by_email(self, email: str) -> Optional[User]:
...
Implémentation des Adapteurs
from pydantic import BaseModel, EmailStr
# Adapteur d'entrée (API REST)
from fastapi import FastAPI, Depends, HTTPException
class CreateUserRequest(BaseModel):
email: EmailStr
name: str
class CreateUserResponse(BaseModel):
id: str
email: str
name: str
app = FastAPI()
@app.post("/users", response_model=CreateUserResponse)
def create_user(
request: CreateUserRequest,
user_service: UserService = Depends(get_user_service)
):
try:
user = user_service.register_user(
email=request.email,
name=request.name
)
return CreateUserResponse(
id=str(user.id),
email=user.email,
name=user.name
)
except UserAlreadyExistsError as e:
raise HTTPException(status_code=400, detail=str(e))
# Adapteur de sortie (Base de données)
# Déjà implémenté comme SQLAlchemyUserRepository
Modèles de conception orientés domaine
Objets de valeur
Objets immuables définis par leurs attributs :
from dataclasses import dataclass
from typing import Pattern
import re
@dataclass(frozen=True)
class Email:
value: str
EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
def __post_init__(self):
if not self.EMAIL_PATTERN.match(self.value):
raise ValueError(f"Email invalide : {self.value}")
def __str__(self):
return self.value
@dataclass(frozen=True)
class Money:
amount: float
currency: str
def __post_init__(self):
if self.amount < 0:
raise ValueError("Le montant ne peut pas être négatif")
if self.currency not in ['USD', 'EUR', 'GBP']:
raise ValueError(f"Devise non prise en charge : {self.currency}")
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Impossible d'ajouter des devises différentes")
return Money(self.amount + other.amount, self.currency)
Agrégats
Groupe d’objets métier traités comme une unité unique :
from dataclasses import dataclass, field
from typing import List
from datetime import datetime
@dataclass
class OrderItem:
product_id: UUID
quantity: int
price: Money
def total(self) -> Money:
return Money(
self.price.amount * self.quantity,
self.price.currency
)
@dataclass
class Order:
id: UUID
customer_id: UUID
items: List[OrderItem] = field(default_factory=list)
status: str = "pending"
created_at: datetime = field(default_factory=datetime.now)
def add_item(self, product_id: UUID, quantity: int, price: Money):
item = OrderItem(product_id, quantity, price)
self.items.append(item)
def remove_item(self, product_id: UUID):
self.items = [
item for item in self.items
if item.product_id != product_id
]
def total(self) -> Money:
if not self.items:
return Money(0, "USD")
return sum(
(item.total() for item in self.items),
Money(0, self.items[0].price.currency)
)
def confirm(self):
if not self.items:
raise ValueError("Impossible de confirmer une commande vide")
if self.status != "pending":
raise ValueError("La commande a déjà été traitée")
self.status = "confirmed"
Événements de domaine
Les événements de domaine permettent un couplage faible entre les composants et supportent les architectures événementielles. Pour des systèmes événementiels à grande échelle, envisagez d’implémenter le streaming d’événements avec des services comme AWS Kinesis—consultez Construire des microservices événementiels avec AWS Kinesis pour un guide détaillé.
from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable
@dataclass
class DomainEvent:
occurred_at: datetime = field(default_factory=datetime.now)
@dataclass
class OrderConfirmed(DomainEvent):
order_id: UUID
customer_id: UUID
total: Money
class EventPublisher:
def __init__(self):
self._handlers: Dict[Type, List[Callable]] = {}
def subscribe(self, event_type: Type, handler: Callable):
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: DomainEvent):
event_type = type(event)
handlers = self._handlers.get(event_type, [])
for handler in handlers:
handler(event)
Fonctionnalités modernes de Python pour une architecture propre
Les fonctionnalités modernes de Python rendent la mise en œuvre d’une architecture propre plus élégante et plus sûre en termes de types. Si vous avez besoin d’un rappel rapide sur la syntaxe et les fonctionnalités de Python, consultez le Python Cheatsheet.
Hints de type et protocoles
from typing import Protocol, runtime_checkable
@runtime_checkable
class Serializable(Protocol):
def to_dict(self) -> dict:
...
@classmethod
def from_dict(cls, data: dict) -> 'Serializable':
...
def serialize(obj: Serializable) -> dict:
return obj.to_dict()
Pydantic pour la validation
from pydantic import BaseModel, Field, validator
from typing import Optional
class CreateUserDTO(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
@validator('name')
def name_must_not_contain_numbers(cls, v):
if any(char.isdigit() for char in v):
raise ValueError('Le nom ne peut pas contenir de chiffres')
return v
class Config:
frozen = True # Rendre immuable
Async/Await pour les opérations d’E/S
La syntaxe async/await de Python est particulièrement puissante pour les opérations liées à l’E/S dans une architecture propre, permettant des interactions non bloquantes avec les bases de données et les services externes. Lors du déploiement d’applications Python sur des plateformes serverless, la compréhension des caractéristiques de performance devient cruciale — consultez AWS lambda performance: JavaScript vs Python vs Golang pour des informations sur l’optimisation des fonctions serverless Python.
from typing import List
import asyncio
class AsyncUserRepository(ABC):
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
class AsyncUserService:
def __init__(self, repository: AsyncUserRepository):
self.repository = repository
async def register_user(self, email: str, name: str) -> User:
user = User(id=uuid4(), email=email, name=name)
return await self.repository.save(user)
async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
tasks = [self.repository.get_by_id(uid) for uid in user_ids]
results = await asyncio.gather(*tasks)
return [u for u in results if u is not None]
Bonnes pratiques pour la structure de projet
Une organisation correcte du projet est essentielle pour maintenir une architecture propre. Avant de configurer la structure de votre projet, assurez-vous d’utiliser des environnements virtuels Python pour l’isolation des dépendances. Le venv Cheatsheet couvre tout ce que vous devez savoir sur la gestion des environnements virtuels. Pour les projets Python modernes, envisagez d’utiliser uv - Nouveau gestionnaire de paquets, de projets et d’environnements Python, qui offre une gestion plus rapide des paquets et une configuration de projet plus simple.
my_application/
├── domain/ # Règles métier de l'entreprise
│ ├── __init__.py
│ ├── entities/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── order.py
│ ├── value_objects/
│ │ ├── __init__.py
│ │ ├── email.py
│ │ └── money.py
│ ├── events/
│ │ ├── __init__.py
│ │ └── user_events.py
│ └── exceptions.py
│
├── application/ # Règles métier de l'application
│ ├── __init__.py
│ ├── use_cases/
│ │ ├── __init__.py
│ │ ├── create_user.py
│ │ └── place_order.py
│ ├── services/
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── ports/
│ ├── __init__.py
│ ├── repositories.py
│ └── external_services.py
│
├── infrastructure/ # Interfaces externes
│ ├── __init__.py
│ ├── persistence/
│ │ ├── __init__.py
│ │ ├── sqlalchemy/
│ │ │ ├── models.py
│ │ │ └── repositories.py
│ │ └── mongodb/
│ │ └── repositories.py
│ ├── messaging/
│ │ ├── __init__.py
│ │ └── rabbitmq_publisher.py
│ ├── external_services/
│ │ ├── __init__.py
│ │ └── email_service.py
│ └── config.py
│
├── presentation/ # Couche UI/API
│ ├── __init__.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── dependencies.py
│ │ ├── routes/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── orders.py
│ │ └── schemas/
│ │ ├── __init__.py
│ │ └── user_schemas.py
│ └── cli/
│ └── commands.py
│
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
│
├── main.py # Point d'entrée de l'application
├── container.py # Configuration de l'injection de dépendances
├── pyproject.toml
└── README.md
Test de l’architecture propre
Test unitaire de la logique du domaine
import pytest
from uuid import uuid4
def test_user_creation():
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
assert user.email == "test@example.com"
assert user.is_active is True
def test_order_total_calculation():
order = Order(id=uuid4(), customer_id=uuid4())
order.add_item(
uuid4(),
quantity=2,
price=Money(10.0, "USD")
)
order.add_item(
uuid4(),
quantity=1,
price=Money(5.0, "USD")
)
assert order.total().amount == 25.0
Test d’intégration avec le dépôt
@pytest.fixture
def in_memory_repository():
return InMemoryUserRepository()
def test_user_repository_save_and_retrieve(in_memory_repository):
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
saved_user = in_memory_repository.save(user)
retrieved_user = in_memory_repository.get_by_id(user.id)
assert retrieved_user is not None
assert retrieved_user.email == user.email
Test de la couche service
from unittest.mock import Mock
def test_user_registration():
# Arrange
mock_repository = Mock(spec=UserRepository)
mock_repository.get_by_email.return_value = None
mock_repository.save.return_value = User(
id=uuid4(),
email="test@example.com",
name="Test"
)
mock_email = Mock(spec=EmailSender)
mock_events = Mock(spec=EventPublisher)
service = UserService(mock_repository, mock_email, mock_events)
# Act
user = service.register_user("test@example.com", "Test")
# Assert
assert user.email == "test@example.com"
mock_repository.save.assert_called_once()
mock_email.send.assert_called_once()
mock_events.publish.assert_called_once()
Pièges courants et comment les éviter
Sur-ingénierie
Ne mettez pas en œuvre une architecture propre pour des applications CRUD simples. Commencez simplement et refactorisez à mesure que la complexité augmente.
Abstractions fuitantes
Assurez-vous que les entités du domaine ne contiennent pas d’annotations de base de données ou de code spécifique à un framework :
# Mauvais
from sqlalchemy import Column
@dataclass
class User:
id: Column(Integer, primary_key=True) # Fuitage de framework vers le domaine
# Bon
@dataclass
class User:
id: UUID # Objet de domaine pur
Dépendances circulaires
Utilisez l’injection de dépendances et les interfaces pour briser les dépendances circulaires entre les couches.
Ignorer le contexte
L’architecture propre n’est pas une solution universelle. Ajustez la rigueur des couches en fonction de la taille du projet et de l’expertise de l’équipe.
Liens utiles
- Clean Architecture par Robert C. Martin
- Documentation sur les hints de type Python
- Documentation Pydantic
- Documentation officielle de FastAPI
- Documentation ORM SQLAlchemy
- Bibliothèque Dependency Injector
- Référence sur le Design Orienté Domaine
- Architecture Patterns with Python
- Blog de Martin Fowler sur l’architecture
- Guide des modèles de conception Python
- Python Cheatsheet
- venv Cheatsheet
- uv - Nouveau gestionnaire de paquets, de projets et d’environnements Python
- Performance d’AWS Lambda : JavaScript vs Python vs Golang
- Conception de microservices événementiels avec AWS Kinesis