pipestage

from pipestage import stream

await (
    stream(documents)
    .flat_map(chunk,  concurrency=8)
    .map(embed,       concurrency=16)
    .batch(100)
    .for_each(upsert, concurrency=4)
)

About

pipestage is an open-source Python library for building async data pipelines with safe staged processing and bounded concurrency. It is developed by OpenLabX and built entirely on the Python standard library with zero runtime dependencies.

The key difference from raw asyncio.gather(): gather is a barrier - stage 2 cannot start until every item from stage 1 is done. pipestage stages are lazy async generators that overlap in time, so stage 2 starts consuming stage 1's output as soon as the first item is ready.

pipestage is aimed at crawlers, ingestion jobs, API fan-out, file processing, and LLM data pipelines.

Table of Contents

• About
• Features
• Install
• Quick Start
• API
  • stream()
  • .map()
  • .filter()
  • .flat_map()
  • .batch()
  • .take()
  • .collect()
  • .for_each()
  • .first()
  • .last()
  • .count()
  • .reduce()
  • .any()
  • .all()
  • Async iteration
  • aclose()
• Error Handling
• Examples
• Project Structure
  • Architecture
  • Concurrency Model
• Dependencies
• Python Versions Tested
• Source Code Version 0.3.0
• Known Issues at v0.3.0
• Contributing
• License
• Contact

Features

• Stage Overlap: All stages run simultaneously. No gather barriers between steps.
• Bounded Concurrency: Per-stage semaphore. Set concurrency once, the rest is handled.
• Ordered or Unordered: Preserve input order, or emit results as ready. One flag.
• Sync and Async: Pass sync or async functions anywhere. No wrapping required.
• Async Generator Source: stream() accepts any async iterable directly.
• Fail-Fast Error Handling: First exception cancels in-flight tasks and propagates unchanged.
• Bounded Memory: Sliding window keeps at most concurrency * 2 tasks alive at once, even on huge or infinite sources.
• Clean Early Exit: aclose() cancels in-flight tasks instead of leaving them running.
• Zero Dependencies: Pure Python standard library. Nothing to install except pipestage itself.

Install

From PyPI:

pip install pipestage

From source:

git clone https://github.com/openlab-x/pipestage.git
cd pipestage
pip install -e .

Requires Python 3.11 or later. No runtime dependencies.

Quick Start

Concurrent fetch and parse:

from pipestage import stream

results = await (
    stream(urls)
    .map(fetch, concurrency=20)
    .filter(lambda r: r["status"] == 200)
    .map(parse, concurrency=8)
    .collect()
)

Batch DB inserts:

await (
    stream(records)
    .map(transform, concurrency=16)
    .batch(100)
    .for_each(insert_batch, concurrency=4)
)

LLM fan-out - emit as ready:

results = await (
    stream(prompts)
    .map(call_llm, concurrency=8, ordered=False)
    .collect()
)

Async generator source:

results = await (
    stream(scan_directory())
    .map(read_file, concurrency=10)
    .filter(lambda f: f["word_count"] >= 500)
    .collect()
)

API

stream()

Create a pipeline from any sync or async iterable.

stream([1, 2, 3])
stream(range(1000))
stream(async_generator())

.map()

.map(fn, *, concurrency=1, ordered=True)

Apply fn to every item. fn may be sync or async. ordered=False emits results as they finish instead of preserving input order.

.filter()

.filter(pred, *, concurrency=1, ordered=True)

Keep only items for which pred returns truthy.

.flat_map()

.flat_map(fn, *, concurrency=1, ordered=True)

Map fn over each item then flatten one level. fn should return a list, generator, or async iterable.

.batch()

.batch(size)

Group items into lists of at most size elements. The final batch may be smaller.

.take()

.take(n)

Stop pulling from the source after n items. Lazy and chainable like map/filter. Cancels any in-flight tasks in an upstream concurrent stage once n is reached.

first_three = await stream(urls).map(fetch, concurrency=20).take(3).collect()

.collect()

Terminal. Consume the pipeline and return all results as a list.

results = await stream(items).map(fn).collect()

.for_each()

.for_each(fn, *, concurrency=1, ordered=True)

Terminal. Consume the pipeline calling fn on each item for side effects. Return values are discarded.

.first()

Terminal. Return the first item, or raise StopAsyncIteration if the source is empty. Cancels any in-flight tasks in an upstream concurrent stage once the first item arrives.

item = await stream(urls).map(fetch, concurrency=20).first()

.last()

Terminal. Consume the whole source and return the final item, or raise StopAsyncIteration if empty.

.count()

Terminal. Consume the whole source and return the number of items, without building a list.

.reduce()

.reduce(fn, initial)

Terminal. Fold the stream left-to-right with fn(acc, item). fn may be sync or async.

total = await stream(orders).map(get_amount, concurrency=8).reduce(lambda acc, x: acc + x, 0)

.any()

.any(pred)

Terminal. Return True on the first item for which pred is truthy, False if none match. Cancels remaining in-flight tasks once a match is found.

.all()

.all(pred)

Terminal. Return False on the first item for which pred is falsy, True if every item matches. Cancels remaining in-flight tasks once a mismatch is found.

Async iteration

Stream is an async iterable. Use it directly without collect().

async for item in stream(records).map(transform, concurrency=8):
    print(item)

aclose()

Close the pipeline and cancel any in-flight tasks. Useful when breaking out of an async for loop early so abandoned work doesn't keep running.

s = stream(urls).map(fetch, concurrency=20)
async for item in s:
    if item["ok"]:
        break
await s.aclose()  # cancels the rest of the in-flight fetches

Error Handling

The pipeline fails fast by default. The first exception stops the pipeline, cancels in-flight tasks, and propagates the original exception unchanged to the caller.

try:
    await stream(urls).map(fetch, concurrency=10).collect()
except RuntimeError as e:
    print(e)

To continue past individual failures, handle exceptions inside fn:

async def safe_fetch(url):
    try:
        return await fetch(url)
    except Exception:
        return None

results = await (
    stream(urls)
    .map(safe_fetch, concurrency=10)
    .filter(lambda r: r is not None)
    .collect()
)

Examples

Each feature in examples/ has three files: a raw asyncio implementation, a pipestage implementation, and a compare script that runs both and prints timing and line-count metrics.

python examples/compare_embed.py   # single comparison
python examples/run_all.py         # all eight

Project Structure

src/pipestage/
    __init__.py     - public entry point: stream()
    _stream.py      - Stream class, fluent API
    _ops.py         - async generator stages
    _utils.py       - internal helpers
tests/
    test_basic.py
    test_concurrency.py
    test_errors.py
    test_lifecycle.py
    test_terminals.py
examples/
    raw_X.py        - plain asyncio implementations
    ps_X.py         - pipestage implementations
    compare_X.py    - timing comparisons

Architecture

Every transformation returns a new Stream wrapping an async generator. Nothing executes until collect() or for_each() is awaited. Each stage pulls from the previous one - all stages run simultaneously.

Concurrency Model

Dependencies

Runtime: none.

Development:

pip install -e ".[dev]"
# installs: pytest, pytest-asyncio, pytest-cov, ruff, mypy

Python Versions Tested

• Python 3.11
• Python 3.12
• Python 3.13
• Python 3.14

Source Code Version 0.3.0

• Core pipeline: stream(), map, filter, flat_map, batch, collect, for_each
• Terminal shortcuts: take, first, last, count, reduce, any, all
• Concurrent execution: ordered and unordered modes with asyncio.Semaphore
• Bounded task creation: sliding window, max concurrency * 2 live tasks, refilled as results are consumed
• Async iteration: Stream usable directly in async for loops
• aclose(): cancels in-flight tasks on early exit; take/first/any/all cancel automatically on short-circuit
• Full test suite: 72 tests across correctness, concurrency, lifecycle, terminals, and error propagation
• Zero runtime dependencies: Python 3.11+ standard library only

Known Issues at v0.3.0

• No per-item timeout. A hung fn call blocks its concurrency slot indefinitely.

Contributing

We welcome contributions.

1. Give the project a star.
2. Follow us on GitHub.
3. Fork the repository.
4. Create a new branch for your feature or fix.
5. Make your changes and add tests.
6. Submit a pull request.

License

This project is licensed under the MIT License.

Contact

In pursuit of innovation,
OpenLabX Team

• Website: https://openlabx.com
• Email: contact@openlabx.com

Follow Us: