Filtering
Prune your stream without breaking the pipeline
.filter(fn) only passes items where fn(x) is truthy; .filter() with no argument drops None values. .filter has no count parameter: to parallelize an expensive predicate (network call, ML inference), compute it in .task(fn, count=N) returning the item or None, then chain .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) # keep only score >= 0.8
.task(enrich, count=4) # parallel enrichment
.wait_for_result()
)⚡ Performance
10 000 items, network predicate (2ms/item)
Sequential 20.0s
4 workers 5.3s
8 workers 2.8s
🚀 7.1× faster
How it works
The FilterPipe consumes items, evaluates the predicate, and forwards only kept items to the next queue. Filtered items are dropped without stalling the stream.