Workflow composition, failure handlers, and nodes

Flytekit workflows are declarative structures that define a directed acyclic graph (DAG) of tasks and other workflows. Unlike tasks, which execute their logic at runtime, the body of a function decorated with @workflow is evaluated at compilation time to build this graph.

Workflow Composition and Promises

When you call a task or a sub-workflow inside a workflow, it does not return the actual value (like an int or str). Instead, it returns a Promise object.

from flytekit import task, workflow

@task
def add_one(x: int) -> int:
    return x + 1

@workflow
def my_workflow(val: int) -> int:
    # result is a Promise, not an int
    result = add_one(x=val)
    return result

The Promise class in flytekit.core.promise acts as a placeholder for a future value. This allows flytekit to track data dependencies between nodes. Because these are not real values during compilation, you cannot use them in standard Python control flow like if result > 0: or for i in range(result):. For conditional logic based on task outputs, you must use flytekit.conditional.

Handling Multiple Outputs

If a task returns multiple values (e.g., using typing.NamedTuple), the call returns a tuple-like object of Promise instances. You can access individual outputs by name or index.

import typing
from flytekit import task, workflow

Outputs = typing.NamedTuple("Outputs", [("sum", int), ("diff", int)])

@task
def math_ops(a: int, b: int) -> Outputs:
    return Outputs(sum=a + b, diff=a - b)

@workflow
def composition_wf(a: int, b: int) -> int:
    res = math_ops(a=a, b=b)
    # Accessing named outputs from the promise tuple
    return res.sum

Explicit Node Creation

While calling tasks directly is the standard way to build workflows, flytekit.core.node_creation.create_node provides lower-level control. This is useful for:

1. Ordering without data flow: Using the >> operator or runs_before method to ensure one node finishes before another starts, even if they don't share data.
2. Dynamic output access: Accessing outputs when the names might be determined programmatically.

from flytekit import task, workflow
from flytekit.core.node_creation import create_node

@task
def t1(): ...

@task
def t2(): ...

@workflow
def ordering_wf():
    node1 = create_node(t1)
    node2 = create_node(t2)
    
    # Ensure t1 runs before t2
    node1 >> node2 
    # Equivalent to: node1.runs_before(node2)

Accessing Node Outputs

When using create_node, the returned object is a Node (from flytekit.core.node). Unlike a standard task call which returns a Promise, a Node object stores its outputs in a .outputs dictionary and also exposes them as attributes (e.g., .o0, .o1).

@task
def produce_val() -> int:
    return 42

@task
def consume_val(x: int): ...

@workflow
def node_output_wf():
    producer = create_node(produce_val)
    
    # Accessing the first output 'o0'
    consume_val(x=producer.o0)
    # Or via the dictionary
    consume_val(x=producer.outputs["o0"])

Per-Node Overrides

You can override resource requirements, timeouts, and retries for specific task executions within a workflow using the .with_overrides() method. This method is available on both Promise objects (returned by task calls) and Node objects (returned by create_node).

from datetime import timedelta
from flytekit import Resources

@workflow
def override_wf(val: int):
    # Overriding on a Promise
    res = add_one(x=val).with_overrides(
        requests=Resources(cpu="2", mem="200Mi"),
        timeout=timedelta(minutes=5),
        retries=3
    )
    
    # Overriding on a Node
    node = create_node(add_one, x=res).with_overrides(node_name="custom-node-name")

The Node.with_overrides method internally updates the NodeMetadata and resource specifications. Note that node_name overrides must be DNS-compliant as they are used for Kubernetes object naming.

Failure Handlers

Flytekit allows you to define a cleanup or notification task that runs if a workflow fails. This is configured via the on_failure parameter in the @workflow decorator.

A failure handler must:

1. Accept all inputs that the workflow itself accepts.
2. Optionally accept a keyword argument err of type flytekit.types.error.error.FlyteError.

import typing
from flytekit import task, workflow
from flytekit.types.error.error import FlyteError

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    if err:
        print(f"Workflow for {name} failed with error: {err.message}")
    else:
        print(f"Workflow for {name} failed for unknown reasons.")

@workflow(on_failure=clean_up)
def my_important_wf(name: str):
    # If any task here fails, clean_up(name=name, err=...) is invoked
    ...

When a failure occurs, Flyte captures the error details and passes them to the on_failure entity. The FlyteError object contains the message and the failed_node_id that triggered the failure.