Workflow composition, failure handlers, and nodes

Workflow composition in flytekit relies on capturing the relationships between tasks, workflows, and launch plans. While simple workflows use direct task calls, complex scenarios require explicit node management, per-node resource overrides, and robust failure handling.

Workflow Composition and Task Outputs

When you define a workflow using the @workflow decorator, calling a task inside the function body does not execute the task immediately. Instead, it returns a Promise object.

Promises and Data Dependencies

A Promise (defined in flytekit.core.promise.Promise) acts as a placeholder for a future value. When you pass a Promise from one task as an input to another, flytekit automatically creates a data dependency between the underlying nodes.

from flytekit import task, workflow

@task
def get_data() -> str:
    return "hello"

@task
def process_data(val: str) -> str:
    return f"{val} world"

@workflow
def my_wf() -> str:
    # data_promise is a Promise object
    data_promise = get_data()
    # Passing the promise creates a dependency: get_data -> process_data
    return process_data(val=data_promise)

Internally, the Promise holds a NodeOutput reference which points back to the Node that produces the value. If a task returns nothing, it returns a VoidPromise, which cannot be used as an input but can still be used to enforce execution order.

Explicit Node Creation

In some cases, you may need to define dependencies between tasks that do not share data, or you may need to programmatically access node attributes. The create_node function in flytekit.core.node_creation allows you to explicitly wrap a task, workflow, or launch plan in a Node.

Using create_node

Unlike a direct task call which returns a Promise, create_node returns a Node object (or a VoidPromise if the entity has no outputs).

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

@task
def t1(): ...

@task
def t2(): ...

@workflow
def manual_wf():
    node_1 = create_node(t1)
    node_2 = create_node(t2)
    
    # Explicitly set execution order using the shift operator
    node_1 >> node_2  # node_1 runs before node_2

Accessing Node Outputs

When using create_node, outputs are not returned directly. Instead, they are attached to the Node object. You can access them via attributes (e.g., .o0, .o1) or through the .outputs dictionary.

@task
def t_output() -> (int, str):
    return 1, "a"

@workflow
def output_access_wf():
    node = create_node(t_output)
    
    # Accessing outputs by attribute
    val_int = node.o0
    val_str = node.o1
    
    # Accessing via the outputs dictionary
    # node.outputs is populated during compilation in create_node
    val_int_alt = node.outputs["o0"]

Per-Node Overrides

Both Promise and Node objects provide a with_overrides method. This allows you to customize the execution parameters of a specific task instance without changing the task definition itself.

The Node.with_overrides method (and its proxy in Promise.with_overrides) supports several parameters:

• requests and limits: Specify Resources (CPU, Memory, etc.).
• timeout: A datetime.timedelta or integer seconds.
• retries: Number of retry attempts.
• interruptible: Boolean for spot/preemptible instance usage.
• container_image: Override the default image for this specific node.

from flytekit import Resources

@workflow
def override_wf(val: int):
    # Overriding on a Promise (returned by task call)
    t1(val=val).with_overrides(
        retries=3,
        requests=Resources(cpu="2", mem="500Mi"),
        node_name="custom-t1-node"
    )
    
    # Overriding on a Node (returned by create_node)
    node = create_node(t2).with_overrides(timeout=600)

Failure Handlers

The @workflow decorator supports an on_failure parameter to define a cleanup or notification task that runs if the workflow fails.

Signature Requirements

A failure handler must follow strict signature rules:

1. It must accept all inputs that the main workflow accepts.
2. It can optionally accept a flytekit.models.core.errors.FlyteError (often typed as typing.Optional[FlyteError]) to inspect the failure.
3. Any additional arguments must be Optional and have default values.

import typing
from flytekit import task, workflow
from flytekit.models.core.errors import FlyteError

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

@task
def failing_task(val: str):
    raise ValueError("Intentional failure")

@workflow(on_failure=clean_up)
def failure_wf(wf_input: str):
    failing_task(val=wf_input)

In this example, clean_up is triggered if failing_task fails. Because failure_wf takes wf_input, the clean_up task must also accept wf_input. Flytekit automatically binds the workflow's inputs to the failure handler at runtime.