Registry / data / pypeln

pypeln

JSON →
library0.4.9pypypi✓ verified 24d ago

PyPeln (pronounced "pypeline") is a Python library designed for building concurrent data pipelines with ease. It offers a simple, functional API that supports multiprocessing (processes), multithreading (threads), and asynchronous programming (asyncio tasks) with the same interface. This allows developers to create multi-stage pipelines and maintain fine-grained control over computational resources. The library is currently at version 0.4.9 and provides solutions for medium-scale data tasks where frameworks like Spark or Dask might be considered overkill.

pip install pypeln
INSTALL
IMPORT
SIG · PYPELN
P
pypeln
datapythonv0.4.9
Install
2.5s avg
Import
565ms
Disk
18MB
Pass rate
10/ 10
Env Coverage10 / 10
glibc
3.93.13
musl
3.93.13
Install & Compatibility
Where this runs
tested against v0.4.9 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
musl
py 3.103.95 runs
installs and imports cleanly · install 0.0s · import 0.364s · 20.6MB
glibc
py 3.103.95 runs
installs and imports cleanly · install 2.5s · import 0.314s · 21MB
18MB installed
● package 18MB
Code
Verified usage

Verified import paths — ran on the pinned version, not inferred.

pypeln
import pypeln
import pypeln

This quickstart demonstrates a basic multiprocessing pipeline. It defines a source iterable (`data`), then chains `map` and `filter` operations using `pl.process`. The pipeline squares numbers, filters for even ones, and prints each processed item. The `workers` and `maxsize` parameters control concurrency and backpressure for each stage. The `list()` call at the end triggers the execution and collects results.

import pypeln as pl import time import random def slow_square(x): time.sleep(random.uniform(0.1, 0.5)) # Simulate work return x * x def is_even(x): return x % 2 == 0 def print_item(x): print(f"Processing: {x}") return x data = range(10) # [0, 1, ..., 9] # Build a multiprocessing pipeline # 1. Map: square each number (4 workers) # 2. Filter: keep only even numbers (2 workers) # 3. Map: print the item (1 worker, ordered output) results = (pl.process.map(slow_square, data, workers=4, maxsize=4) .filter(is_even, workers=2, maxsize=2) .map(print_item, workers=1, maxsize=1, ordered=True)) # Consume the results (this starts the pipeline execution) final_list = list(results) print(f"Final result: {final_list}")
Debug
Known issues
breakingThe `maxsize` argument was removed from all `from_iterable` functions in version 0.4.0.
fix
Review `from_iterable` calls and remove the `maxsize` argument if present. The functionality for `maxsize` was later introduced for `to_stage` and `to_iterable`, and `ordered` functions in 0.4.6.
affects: >=0.4.0
gotchaPython 3.6 versions between 0.4.0 and 0.4.1 experienced import errors due to reliance on `typing.Protocol` (introduced in Python 3.8). Although the library generally targets Python 3.6+, the `typing.Protocol` dependency broke compatibility in these specific versions.
fix
Upgrade to PyPeln version 0.4.2 or newer, which includes conditional dependencies and imports to correctly support Python >= 3.6.2. Ensure `typing_extensions` is installed.
affects: 0.4.0 - 0.4.1
gotchaThe `task` module, which provides `asyncio` based pipelines, is only available and fully supported for Python versions 3.7 and above, despite the overall library supporting Python >= 3.6.2.
fix
When using `pl.task` for asynchronous operations, ensure your Python environment is 3.7 or newer. For older Python 3.6 environments, use `pl.process` or `pl.thread` instead.
affects: <3.7 for `task` module
gotchaThe `maxsize` parameter in `pl.process.map` and `pl.thread.map` was not correctly respected, potentially leading to unbounded queues and out-of-memory issues in earlier versions.
fix
Upgrade to PyPeln version 0.4.7 or newer to ensure `maxsize` properly functions as a backpressure mechanism, preventing upstream stages from overwhelming downstream ones.
affects: <0.4.7
gotchaProcess workers using multiprocessing start method 'spawn' could raise an `AttributeError` on certain systems.
fix
Upgrade to PyPeln version 0.4.9. This issue was specifically addressed in this release.
affects: <0.4.9
Upgrade
Version history
0.4.9latest on PyPI · released Jan 6, 2022
Audit
Dependencies
pythonrequiredRequired Python version compatibility.
stopitrequiredRuntime dependency for managing worker lifecycle and timeouts.
typing_extensionsrequiredProvides backports of typing features for older Python versions, especially important for Python < 3.8.
Agent activity
31 hits · last 30 days
node
26
OpenAI (training)
1
Resources
pypeln — pip install pypeln · libregistry