Tâches parallèles
Multipliez les workers d'un seul paramètre
.task(fn, count=N) exécute fn dans N processus séparés qui partagent la même file d'entrée. Chaque worker tourne dans son propre processus, contournant le GIL Python — idéal pour les traitements CPU-bound et les appels I/O.
Signature
Pipeline.task(task, count=1, on_error=None) -> Pipeline[OutPacket]from olympipe import Pipeline
import time
def slow_compute(x: int) -> int:
time.sleep(0.01) # simule un calcul lourd
return x ** 2
# 1 worker séquentiel
results = Pipeline(range(100)).task(slow_compute).wait_for_result()
# 8 workers parallèles — ~8x plus rapide
results = (
Pipeline(range(100))
.task(slow_compute, count=8)
.wait_for_result()
)⚡ Performance
100 items, calcul CPU (0.01s/item)
Comment ça fonctionne
Olympipe crée N processus enfants via multiprocessing. Chaque item de la source est dépilé par le premier worker disponible — pas de coordination manuelle, pas de Lock, pas de Pool.
on_error : gérer les erreurs item par item
Signature : .task(task, count=1, on_error=None). Si fn lève une exception, on_error(item, exc) est appelé dans le worker : renvoyer None abandonne l'item, toute autre valeur le remplace. Sans on_error, l'erreur fait échouer le pipeline (PipelineExecutionError).
from olympipe import Pipeline
def parse(s: str) -> int:
return int(s)
def fallback(item: str, exc: Exception):
print(f"skip {item!r}: {exc}")
return None # None = abandonner l'item, toute autre valeur le remplace
print(Pipeline(["1", "2", "oops", "4"]).task(parse, on_error=fallback).wait_for_result())
print(Pipeline(["1", "x"]).task(parse, on_error=lambda item, exc: -1).wait_for_result())skip 'oops': invalid literal for int() with base 10: 'oops'
[1, 2, 4]
[1, -1]Exemples liés
Redimensionnement d'images en masse
Redimensionner 10 000 images en parallèle avec Pillow.
Scraping web parallèle
Télécharger et parser 1 000 pages HTML avec httpx.
Traitement de fichiers CSV
Lire et transformer des milliers de fichiers CSV en parallèle.