Pipeline tasks

Chain tasks left-to-right with pipeline(*tasks, input=...), passing each step's output directly into the next.

pipeline from flux.tasks runs a sequence of tasks one after another, threading the output of each step into the input of the next. Use it when your work is a linear chain of transformations: each step depends on the step before it and produces a single value for the step after.

Basic usage

Import pipeline and pass task functions — not coroutines — along with the input keyword argument:

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: int) -> int:
    return x * 2


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


@task
async def square(x: int) -> int:
    return x * x


@workflow
async def math_pipeline(ctx: ExecutionContext[int]):
    result = await pipeline(
        multiply_by_two,
        add_three,
        square,
        input=ctx.input,
    )
    return result


if __name__ == "__main__":
    ctx = math_pipeline.run(5)
    print(ctx.output)        # 169  (5→10→13→169)
    print(ctx.has_succeeded) # True

Given input 5, the three steps run in order: 5 × 2 = 10, 10 + 3 = 13, 13² = 169. The input keyword argument is required. It is the value passed to the first task.

A realistic example: ETL pipeline

pipeline fits any extract-transform-load pattern where each stage produces input for the next:

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


@task
async def fetch_raw(record_id: str) -> dict:
    # Replace with a real data source call
    return {"id": record_id, "value": " 42 ", "tags": "a,b,c"}


@task
async def clean(record: dict) -> dict:
    return {k: v.strip() if isinstance(v, str) else v for k, v in record.items()}


@task
async def enrich(record: dict) -> dict:
    return {**record, "tags": record["tags"].split(",")}


@task
async def validate(record: dict) -> dict:
    if not record.get("id"):
        raise ValueError("record must have an id")
    return record


@workflow
async def etl_workflow(ctx: ExecutionContext[str]):
    return await pipeline(fetch_raw, clean, enrich, validate, input=ctx.input)


if __name__ == "__main__":
    ctx = etl_workflow.run("rec-001")
    print(ctx.output)
    # {'id': 'rec-001', 'value': '42', 'tags': ['a', 'b', 'c']}
    print(ctx.has_succeeded)  # True

The four steps run sequentially. The raw record from fetch_raw flows through clean, then enrich, then validate. Each step receives exactly one positional argument: the return value of the previous step.

Durability

Every step in the pipeline is a durable checkpoint. After each task completes, its result is recorded to the execution log. If the workflow is interrupted and replayed — by a worker restart or a crash — Flux restores the already-recorded results and re-runs only the steps that had not yet finished. A pipeline of ten steps interrupted at step seven replays from step seven, not step one.

Error semantics

When any task in the pipeline raises an exception, the pipeline stops immediately and the workflow fails. Steps that come after the failing step never run.

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


@task
async def parse_int(value: str) -> int:
    return int(value)          # raises ValueError for non-numeric input


@task
async def double(n: int) -> int:
    return n * 2


@workflow
async def convert_workflow(ctx: ExecutionContext[str]):
    return await pipeline(parse_int, double, input=ctx.input)


if __name__ == "__main__":
    ctx = convert_workflow.run("not-a-number")
    print(ctx.has_succeeded)  # False
    print(ctx.has_failed)     # True

To recover from a mid-chain failure, attach retry_max_attempts or a fallback to the specific task that may fail. See Errors and retries for the full options.

pipeline vs. sequential awaits

A pipeline and a plain sequence of await calls produce the same behavior and the same durability guarantees:

# Pipeline form
result = await pipeline(normalize, tokenize, count_tokens, input=ctx.input)

# Equivalent sequential form
normalized = await normalize(ctx.input)
tokens     = await tokenize(normalized)
result     = await count_tokens(tokens)

Both checkpoint every step. Use pipeline when the chain is long and the steps are pure data transformations, where the linear declaration avoids accumulating intermediate variable names. Use sequential await calls when you need branching, conditionals, or side effects between steps.

What’s next