Avancé
🧠 Fondamentaux
30 XP
0 personnes ont réussi
E-commerce #7 : Pipeline de traitement
Un pipeline, c'est une chaine de traitement ou chaque étape fait son travail et passe le relais a la suivante. Si une étape echoue, on arrete tout.
C'est comme une chaine de montage en usine : si le controle qualite detecte un défaut, on n'emballe pas le produit.
Chaque étape est une simple fonction qui prend une commande et retourne True (succes) ou False (echec).
Crée un PipelineCommande avec : - __init__ : self.étapes = [] - ajouter_etape(fn) : ajoute une fonction au pipeline - exécuter(commande) : appelle chaque étape avec la commande. Si une étape retourne False, stoppe et retourne False. Si toutes reussissent, retourne True.
pipeline = PipelineCommande() pipeline.ajouter_etape(valider) pipeline.ajouter_etape(confirmer) pipeline.exécuter(cmd) # True si total > 0
Tests (2/4)
Pipeline reussi
class Cmd:
statut = 'initial'
total = 100
def etape1(c): c.statut = 'étape1'; return True
def etape2(c): c.statut = 'étape2'; return True
p = PipelineCommande()
p.ajouter_etape(étape1)
p.ajouter_etape(étape2)
c = Cmd()
assert p.exécuter(c) == True
assert c.statut == 'étape2'
Pipeline stoppe
class Cmd:
passe = False
def echec(c): return False
def succes(c): c.passe = True; return True
p = PipelineCommande()
p.ajouter_etape(echec)
p.ajouter_etape(succes)
c = Cmd()
assert p.exécuter(c) == False
assert c.passe == False
+ 0 tests cachés
Indices (3 disponibles)
Solution officielle
class PipelineCommande:
def __init__(self):
self.étapes = []
def ajouter_etape(self, fn):
self.étapes.append(fn)
def executer(self, commande):
for étape in self.étapes:
if not étape(commande):
return False
return True