

pl.task.run

Iterates over one or more stages until their iterators run out of elements.

import pypeln as pl

data = get_data()
stage = pl.process.each(slow_fn, data, workers=6)

# execute pipeline
pl.process.run(stage)

Parameters:

Name	Type	Description	Default
stages	Union[pypeln.task.stage.Stage[~A], Iterable[~A], AsyncIterable[~A]]	A stage/iterable or list of stages/iterables to be iterated over. If a list is passed, stages are first merged using concat before iterating.	()
maxsize	int	The maximum number of objects the stage can hold simultaneously, if set to 0 (default) then the stage can grow unbounded.	0

Source code in pypeln/task/api/run.py

def run(
    *stages: tp.Union[Stage[A], tp.Iterable[A], tp.AsyncIterable[A]], maxsize: int = 0
) -> None:
    """
    Iterates over one or more stages until their iterators run out of elements.

    ```python
    import pypeln as pl

    data = get_data()
    stage = pl.process.each(slow_fn, data, workers=6)

    # execute pipeline
    pl.process.run(stage)
    ```

    Arguments:
        stages: A stage/iterable or list of stages/iterables to be iterated over. If a list is passed, stages are first merged using `concat` before iterating.
        maxsize: The maximum number of objects the stage can hold simultaneously, if set to `0` (default) then the stage can grow unbounded.

    """

    if len(stages) == 0:
        return
    elif len(stages) == 1:
        stage = to_iterable(stages[0], maxsize=maxsize)
    else:
        stage = concat(list(stages), maxsize=maxsize)

    for _ in stage:
        pass