data / sqlalchemy
Patterns d'architecture : repository et unit of work
Explication
Ce que vous allez apprendre
- Isoler la logique métier de l'accès aux données avec le pattern Repository
- Regrouper repositories et gestion transactionnelle dans une Unit of Work
- Écrire un
InMemoryRepositorypour tester la logique métier sans base de données réelle - Reconnaître quand ces patterns sont justifiés, et quand ils ajoutent une complexité inutile
Dans quel contexte ?
Une équipe qui maintient une application de gestion de comptes bancaires veut tester sa logique de virement (leçon 9) sans démarrer une vraie base PostgreSQL à chaque exécution de test, ce qui ralentit considérablement la suite de tests en CI. En isolant l'accès aux données derrière un repository, elle peut substituer une implémentation en mémoire pure Python pendant les tests, et la vraie implémentation SQLAlchemy en production.
D'abord, un couplage qui devient gênant
Jusqu'ici, le code métier (créer un compte, transférer de l'argent) appelait directement session.add(), select(), etc. Sur un petit projet, c'est parfaitement raisonnable. Mais sur un projet plus gros, cela crée un couplage fort : impossible de tester la logique métier sans une vraie base de données.
Étape 1 : une façade devant les données
Un repository est une interface qui expose des opérations métier, comme "trouve un utilisateur par email", sans exposer les détails de leur implémentation SQL. Le code métier ne sait pas si, derrière cette interface, il y a PostgreSQL ou autre chose — comme brancher un appareil sur une prise standard sans se soucier de la centrale électrique.
| Pattern | Rôle |
|---|---|
| Repository | isole l'accès aux données derrière une interface métier |
| Unit of Work | regroupe repositories + transaction dans une frontière claire |
InMemoryRepository | implémentation de test, sans base de données réelle |
Il reste un problème : où placer la transaction ?
Une fois qu'on isole l'accès aux données derrière un repository, il faut aussi décider où placer le commit et le rollback, sans que chaque service métier ait à s'en soucier directement.
Étape 2 : regrouper repository et transaction
L'Unit of Work résout ce problème en regroupant un ou plusieurs repositories avec la gestion transactionnelle vue en leçon 9, dans un objet unique utilisé comme contexte (with uow: ...). Cela donne une frontière claire à chaque opération métier : soit tout réussit et se valide ensemble, soit rien n'est appliqué.
Le vrai bénéfice de tout ce travail
L'avantage le plus concret est de pouvoir écrire un InMemoryRepository pour les tests, qui simule le stockage en pur Python. Les tests deviennent instantanés et n'exigent aucune base réelle.
Un dernier point d'attention
Ces patterns ajoutent une couche d'indirection : ils se justifient sur des projets avec une logique métier substantielle, pas sur un simple CRUD où ils ajouteraient de la complexité sans bénéfice réel.
Prérequis
Repository et Unit of Work ne sont pas des règles universelles : sur une application CRUD simple, ils ajoutent des fichiers et de l'indirection sans bénéfice proportionnel. Réservez-les à une logique métier suffisamment riche pour justifier ce découplage.
Vers la suite
Une fois l'architecture posée, la prochaine leçon revient à un sujet très concret : comment repérer et corriger les vrais goulots d'étranglement de performance d'une application SQLAlchemy.
Commandes & code
Patterns d'architecture
from abc import ABC, abstractmethod
from sqlalchemy.orm import Session
from sqlalchemy import select
# --- Pattern Repository : isole l'accès aux données de la logique métier ---
class UtilisateurRepository(ABC):
@abstractmethod
def get(self, id: int) -> Utilisateur | None: ...
@abstractmethod
def get_by_email(self, email: str) -> Utilisateur | None: ...
@abstractmethod
def add(self, utilisateur: Utilisateur) -> None: ...
@abstractmethod
def list_actifs(self) -> list[Utilisateur]: ...
class SqlAlchemyUtilisateurRepository(UtilisateurRepository):
def __init__(self, session: Session):
self.session = session
def get(self, id: int) -> Utilisateur | None:
return self.session.get(Utilisateur, id)
def get_by_email(self, email: str) -> Utilisateur | None:
stmt = select(Utilisateur).where(Utilisateur.email == email)
return self.session.scalar(stmt)
def add(self, utilisateur: Utilisateur) -> None:
self.session.add(utilisateur)
def list_actifs(self) -> list[Utilisateur]:
stmt = select(Utilisateur).where(Utilisateur.est_actif == True)
return list(self.session.scalars(stmt).all())
# Repository en mémoire pour les tests unitaires (aucune base de données requise)
class InMemoryUtilisateurRepository(UtilisateurRepository):
def __init__(self):
self._data: dict[int, Utilisateur] = {}
self._next_id = 1
def get(self, id: int) -> Utilisateur | None:
return self._data.get(id)
def get_by_email(self, email: str) -> Utilisateur | None:
return next((u for u in self._data.values() if u.email == email), None)
def add(self, utilisateur: Utilisateur) -> None:
utilisateur.id = self._next_id
self._data[self._next_id] = utilisateur
self._next_id += 1
def list_actifs(self) -> list[Utilisateur]:
return [u for u in self._data.values() if u.est_actif]
# --- Pattern Unit of Work : regroupe repositories + transaction ---
class UnitOfWork:
def __init__(self, session_factory=SessionLocal):
self.session_factory = session_factory
def __enter__(self):
self.session: Session = self.session_factory()
self.utilisateurs = SqlAlchemyUtilisateurRepository(self.session)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
self.session.rollback()
self.session.close()
def commit(self):
self.session.commit()
def rollback(self):
self.session.rollback()
# Utilisation dans un service métier : découplé de SQLAlchemy
def creer_compte_utilisateur(email: str, nom: str, uow: UnitOfWork) -> Utilisateur:
with uow:
if uow.utilisateurs.get_by_email(email) is not None:
raise ValueError("Email déjà utilisé")
utilisateur = Utilisateur(email=email, nom=nom)
uow.utilisateurs.add(utilisateur)
uow.commit()
return utilisateur
# Test unitaire SANS base de données réelle grâce à l'abstraction
def test_creer_compte_rejette_doublon():
class FakeUow:
def __enter__(self):
self.utilisateurs = InMemoryUtilisateurRepository()
return self
def __exit__(self, *a): pass
def commit(self): pass
uow = FakeUow()
with uow:
uow.utilisateurs.add(Utilisateur(email="x@y.com", nom="X"))
try:
creer_compte_utilisateur("x@y.com", "Doublon", uow)
assert False, "aurait dû lever une exception"
except ValueError:
passRésumé
- Le pattern Repository masque SQLAlchemy derrière une interface abstraite, testable sans base réelle.
- Le pattern Unit of Work regroupe repositories et transaction, offrant une frontière claire pour les services métier.
- Ces patterns ont un coût (indirection, code supplémentaire) : pertinents surtout sur de gros projets multi-équipes ou avec logique métier complexe.
- Un
InMemoryRepositoryaccélère radicalement les tests unitaires en évitant tout accès disque/réseau.
Exercices pratiques
Mission : rendre une suite de tests de virement bancaire instantanée
Objectif : Isoler la logique métier de virement derrière un repository pour la tester sans base de données réelle.
Contexte
La suite de tests de la logique de virement (leçon 9) démarre une vraie base PostgreSQL avant chaque test, ce qui ralentit considérablement la CI. L'équipe veut appliquer le pattern Repository et Unit of Work pour substituer une implémentation en mémoire pure Python pendant les tests, sans changer le service métier creer_compte_utilisateur.