pl.process.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.process.stage.Stage[~A], Iterable[~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/process/api/run.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def run(*stages: tp.Union[Stage[A], tp.Iterable[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