pyper Documentation

Pyper unifies synchronous and asynchronous execution. An AsyncPipeline is created whenever you wrap an async def function with task().

When composing pipelines using |, the resulting pipeline type follows these rules:

• Pipeline + Pipeline = Pipeline
• Pipeline + AsyncPipeline = AsyncPipeline
• AsyncPipeline + Pipeline = AsyncPipeline
• AsyncPipeline + AsyncPipeline = AsyncPipeline

Rule of thumb: If a pipeline contains at least one asynchronous task, the entire resulting pipeline becomes an AsyncPipeline.

When using an AsyncPipeline, consumer functions must be able to handle AsyncIterable inputs (e.g., using async for).

import asyncio
import json
from typing import AsyncIterable, Dict
from pyper import task

async def step1(limit: int):
    for i in range(limit):
        yield {"data": i}

def step2(data: Dict):
    return data | {"hello": "world"}

class AsyncJsonFileWriter:
    def __init__(self, filepath):
        self.filepath = filepath
    
    async def __call__(self, data: AsyncIterable[Dict]):
        # Must use 'async for' to consume AsyncIterable
        data_list = [row async for row in data]
        with open(self.filepath, 'w', encoding='utf-8') as f:
            json.dump(data_list, f, indent=4)

async def main():
    run = (
        task(step1, branch=True)
        | task(step2)
        > AsyncJsonFileWriter("data.json")
    )
    await run(limit=10)

if __name__ == "__main__":
    asyncio.run(main())

Generators allow for lazy execution, processing items one by one instead of loading entire datasets into memory.

• branch=True: When used with a generator, Pyper treats each yield as an individual output that is immediately passed to the next task in the pipeline. This is highly efficient for large volumes of data.
• Without branch=True: The task will output the single generator object itself as the result.

Limitations:

• Synchronous Generators in AsyncPipelines: Synchronous generators do not benefit from threading/multiprocessing in an async context because the generator object itself is returned immediately, performing its work outside the worker threads.
• Multiprocessing/Pickling: On Windows, generator objects cannot be pickled. You cannot pass a generator object directly as an input or output to a multiprocess=True task. However, using branch=True to pass individual yielded values is fine.

from pyper import task

def get_data():
    yield 1
    yield 2
    yield 3

if __name__ == "__main__":
    # Each value is yielded to the loop individually
    branched_pipeline = task(get_data, branch=True)
    for output in branched_pipeline():
        print(output)
        #> 1
        #> 2
        #> 3

    # The loop receives a single generator object
    non_branched_pipeline = task(get_data)
    for output in non_branched_pipeline():
        print(output)
        #> <generator object get_data at ...>

To handle complex data flows where you need to process batches of data rather than individual items, you can nest pipelines.

If a pipeline (e.g., download_files_from_source) generates multiple outputs per input (using branch=True), you can wrap that entire pipeline in a task() call. When used inside an outer pipeline, this nested task will treat the entire generator of outputs from the inner pipeline as a single batch (a single output) for the next step in the outer pipeline.

# Inner pipeline: generates multiple outputs per source
download_files_from_source = (
    task(list_files, branch=True)  # Return a list of file info
    | task(download_file, workers=20)  # Return a filepath
    | task(decrypt_file, workers=5, multiprocess=True)  # Return a filepath
)

# Outer pipeline: treats the inner pipeline as a single task that returns a batch
download_and_merge_files = (
    task(get_sources, branch=True)  # Return a list of sources
    | task(download_files_from_source)  # Return a batch of filepaths (as a generator)
    | task(sync_files, workers=5)  # Do something with each batch
)

Pyper supports three modes of execution to handle different types of workloads. Choosing the right one depends on whether your task is IO-bound or CPU-bound:

• Threading (Default): Best for IO-bound tasks (e.g., network requests, database reads, sleeping) where you want moderate overhead and synchronous implementation.
• Multiprocessing (multiprocess=True): Best for CPU-bound tasks (e.g., heavy math, text parsing, sorting). This provides true parallelism by bypassing the GIL, but has high overhead.
• Async (async def): Best for IO-bound tasks with low overhead. Requires using await or yield to allow concurrency. Warning: Do not use blocking calls (like time.sleep()) inside an async def function, as it will block the entire event loop.

# CPU-bound: Use multiprocess=True
def long_computation(data: int):
    for i in range(1, 1_000_000):
        data *= i
    return data

# Correct way for CPU-bound
pipeline = task(long_computation, workers=10, multiprocess=True)

# Incorrect way for CPU-bound (will not benefit from concurrency)
pipeline = task(long_computation, workers=10)

Pyper allows you to compose standard Python functions into a concurrent data flow using the task wrapper and pipe operators. You can define how tasks behave using specific arguments:

• branch=True: Use this when a task yields multiple items (a generator). This allows downstream tasks to process each yielded item individually.
• workers=N: Specifies the number of concurrent workers to run a specific task, useful for I/O-bound operations like HTTP requests.
• bind=task.bind(...): Used to inject specific arguments into a task that are not part of the primary data flow (e.g., a shared session object or a constant configuration value).
• Pipe Operators:
  • | (Pipe): Passes the output of one task as the first argument to the next task.
  • > (Sink): Connects the final stream of data to a terminal function (a 'sink') that consumes the entire collection.

This pattern allows you to build complex, concurrent pipelines with minimal code coupling.

from pyper import task

# Example pipeline construction
run = (
    task(generate_urls_by_month, branch=True)
    | task(
        fetch_text_data,
        workers=3,
        bind=task.bind(session=session))
    | task(
        read_game_data,
        branch=True,
        bind=task.bind(player=player))
    > build_df
)

# Execute the pipeline
df = run(player, num_months)

The join parameter controls how a producer-consumer task receives input from the previous task:

• join=False (default): The task takes each individual output from the previous task as a separate input.
• join=True: The task takes a stream (an Iterable) of inputs from the previous task.

Important Considerations:

• A Producer cannot have join=True.
• If join=True is used with multiple workers, the order of outputs cannot be guaranteed consistently.

from typing import Iterable
from pyper import task

def create_data(x: int):
    return [x + 1, x + 2, x + 3]

def running_total(data: Iterable[int]):
    total = 0
    for item in data:
        total += item
        yield total

if __name__ == "__main__":
    pipeline = (
        task(create_data, branch=True)
        | task(running_total, branch=True, join=True)
    )
    for output in pipeline(0):
        print(output)
        #> 1
        #> 3
        #> 6
    )