Simple Pipeline

Chain tasks end to end with the built-in pipeline helper.

Chains three tasks so each one’s output feeds the next. The pipeline built-in takes a sequence of tasks plus an input and threads the value through them in order. Reach for it when you have a linear transformation and don’t want to hand- wire every intermediate await.

Run it

python examples/simple_pipeline.py

from __future__ import annotations

from flux import ExecutionContext
from flux.task import task
from flux.tasks import pipeline
from flux.workflow import workflow


@task
async def multiply_by_two(x):
    return x * 2


@task
async def add_three(x):
    return x + 3


@task
async def square(x):
    return x * x


@workflow
async def simple_pipeline(ctx: ExecutionContext[int]):
    if not ctx.input:
        raise TypeError("Input not provided")
    result = await pipeline(multiply_by_two, add_three, square, input=ctx.input)
    return result


if __name__ == "__main__":  # pragma: no cover
    ctx = simple_pipeline.run(5)
    print(ctx.to_json())

pipeline(multiply_by_two, add_three, square, input=5) runs multiply_by_two(5), feeds 10 into add_three, then 13 into square, yielding 169. Each step is a recorded task, so a replay resumes from the last completed step rather than restarting the chain.