Workflow composition, failure handlers, and nodes

Flytekit workflows orchestrate the execution of tasks and other workflows by defining a directed acyclic graph (DAG) of nodes. While the @workflow decorator allows you to define these relationships using standard Python function calls, flytekit provides lower-level primitives for explicit node management, dependency control, and error handling.

Workflow Composition and Promises

When you call a task inside a @workflow function, flytekit does not execute the task immediately. Instead, it creates a Node and returns a Promise (defined in flytekit.core.promise.Promise).

A Promise is a placeholder for a value that will be available at runtime. During workflow compilation, these promises are used to track data dependencies between nodes.

from flytekit import task, workflow

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

@task
def process_value(v: int) -> int:
    return v + 1

@workflow
def my_wf() -> int:
    # val is a Promise object, not an int
    val = get_value()
    # Passing the promise to another task creates a data dependency
    return process_value(v=val)

Internally, Promise objects wrap a NodeOutput which points back to the Node that produces the value. If you try to perform Python operations like range(val) or if val > 0: inside the workflow body, flytekit will raise an error because the actual value is not available during compilation.

Explicit Node Creation

In some scenarios, you may need to define execution order without a direct data dependency (e.g., ensuring a setup task runs before a processing task). The create_node function in flytekit.core.node_creation allows you to explicitly instantiate a node.

Dependency Management

You can use the >> operator or the runs_before method on a Node to enforce execution order:

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

@task
def setup():
    print("Setting up...")

@task
def work():
    print("Working...")

@workflow
def manual_dependency_wf():
    setup_node = create_node(setup)
    work_node = create_node(work)
    
    # Ensure setup runs before work
    setup_node >> work_node

Accessing Node Outputs

When using create_node, outputs are accessed via attributes on the node object (e.g., node.o0, node.o1) or through the node.outputs dictionary. This differs from standard task calls which return promises directly.

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

@workflow
def output_access_wf():
    n = create_node(multi_output)
    # Accessing outputs by name (default names are o0, o1, ...)
    use_task(v=n.o0, s=n.o1)

Per-Node Overrides

Flytekit allows you to customize the execution parameters of individual nodes using the with_overrides method. This is available on both Promise objects and Node objects.

Common overrides include:

• requests and limits: Resource requirements using flytekit.Resources.
• timeout: A datetime.timedelta or integer seconds.
• retries: Number of retries on failure.
• interruptible: Boolean indicating if the node can be run on spot instances.
• container_image: Use a specific image for this node.

from flytekit import Resources

@workflow
def override_wf(val: int):
    # Overriding a task call (Promise)
    t1 = task_a(val=val).with_overrides(
        requests=Resources(cpu="2", mem="500Mi"),
        retries=3
    )
    
    # Overriding an explicit node
    node_b = create_node(task_b).with_overrides(
        node_name="custom-node-name",
        timeout=600
    )

The Node.with_overrides method (in flytekit/core/node.py) updates the NodeMetadata and resource specifications for the underlying Flyte entity.

Failure Handlers

Workflows can define an on_failure handler that executes if any node in the workflow fails. This is configured via the @workflow decorator.

Signature Requirements

A failure handler must be a task or workflow that accepts:

1. All inputs of the parent workflow.
2. An additional argument named error which must be Optional.

from flytekit import task, workflow
from flytekit.models.core.errors import Error
from typing import Optional

@task
def clean_up(name: str, error: Optional[Error] = None):
    if error:
        print(f"Workflow {name} failed with error: {error.message}")

@workflow(on_failure=clean_up)
def main_wf(name: str):
    task_that_might_fail(name=name)

The on_failure mechanism ensures that resources can be cleaned up or notifications sent regardless of where the workflow failed. Flytekit enforces that any additional parameters in the failure handler beyond the workflow's own inputs must have default values (typically None).