Filtrage
Élaguez votre flux sans rompre le pipeline
.filter(fn) ne laisse passer que les items pour lesquels fn(x) est vrai ; .filter() sans argument retire les None. .filter n'a pas de paramètre count : pour paralléliser un prédicat coûteux (appel réseau, inférence ML), calculez-le dans .task(fn, count=N) qui renvoie l'item ou None, puis enchaînez .filter().
Signature
Pipeline.filter(keep_if_true=None) -> Pipeline[InPacket]python
from olympipe import Pipeline
def is_valid(record: dict) -> bool:
return record.get("score", 0) >= 0.8 and record.get("label") is not None
results = (
Pipeline(raw_records) # 10 000 records
.filter(is_valid) # garde seulement score >= 0.8
.task(enrich, count=4) # enrichissement parallèle
.wait_for_result()
)⚡ Performance
10 000 items, prédicat réseau (2ms/item)
Séquentiel 20.0s
4 workers 5.3s
8 workers 2.8s
🚀 7.1× plus rapide
Comment ça fonctionne
Le FilterPipe consomme les items, évalue le prédicat et envoie uniquement les items retenus dans la file suivante. Les items filtrés sont abandonnés sans bloquer le flux.