Workflow composition, failure handlers, and nodes

Flytekit workflows are defined by decorating Python functions with @workflow. While these functions look like standard Python, they serve as a DSL for building a directed acyclic graph (DAG) of execution. During compilation, task calls return Promise objects rather than actual values, representing the future output of a node in the graph.

Workflow Composition and Promises

When you call a task inside a workflow, flytekit creates a Node and returns one or more Promise objects. These promises are passed to subsequent tasks to establish data dependencies.

from flytekit import task, workflow

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

@task
def process_value(val: int) -> str:
    return f"Value is {val}"

@workflow
def my_workflow() -> str:
    # 'val_promise' is a flytekit.core.promise.Promise
    val_promise = get_value()
    # Passing the promise to another task creates a data dependency
    return process_value(val=val_promise)

Internally, the Promise class (found in flytekit/core/promise.py) tracks the NodeOutput it refers to. If you attempt to use a promise in a standard Python context—such as range(val_promise) or if val_promise:—flytekit will raise a ValueError because the actual value is not available until execution time on the Flyte cluster.

Explicit Node Creation

In some scenarios, you need to manage execution order without passing data, or you need to apply specific overrides to a single step. The create_node function in flytekit.core.node_creation allows you to manually instantiate a Node.

Ordering with the Shift Operator

If two tasks have no data dependency but must run in a specific order, use the >> operator or the runs_before method on the Node object.

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

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

@task
def compute():
    print("Computing...")

@workflow
def ordered_workflow():
    setup_node = create_node(setup)
    compute_node = create_node(compute)
    
    # Enforce that setup runs before compute
    setup_node >> compute_node

Accessing Node Outputs

Unlike a direct task call which returns a Promise, create_node returns a Node object. To access the outputs of the underlying task for use in other tasks, you must use the .o<index> attribute pattern (e.g., .o0, .o1).

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

@workflow
def node_output_workflow():
    node = create_node(multi_output)
    
    # node.o0 and node.o1 are the Promises for the task's outputs
    other_task(val=node.o0, label=node.o1)

Per-Node Overrides

The Node class provides a with_overrides method to customize execution parameters for a specific instance of a task within a workflow. This is useful for adjusting resources, timeouts, or retries without modifying the task definition itself.

from flytekit import Resources

@workflow
def override_workflow(val: int):
    # Apply overrides to a standard task call promise
    node_promise = get_value().with_overrides(
        requests=Resources(cpu="2", mem="200Mi"),
        retries=3,
        node_name="custom-resource-node"
    )
    
    # Or apply to a node created via create_node
    manual_node = create_node(compute).with_overrides(timeout=600)

The with_overrides method (implemented in flytekit/core/node.py) updates the NodeMetadata and resource requirements. Note that node_name overrides are automatically "dnsified" to comply with Kubernetes naming standards.

Failure Handlers

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

Signature Requirements

A failure handler must follow strict signature rules:

1. It must accept all inputs defined in the parent workflow.
2. It can optionally accept a parameter of type FlyteError (usually named err or error) to inspect the failure details.
3. Any additional parameters must be Optional and have default values.

import typing
from flytekit import task, workflow
from flytekit.exceptions.user import FlyteError

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

@task
def failing_task(name: str):
    raise ValueError("Simulated failure")

@workflow(on_failure=clean_up)
def failure_handling_wf(name: str):
    failing_task(name=name)

When failure_handling_wf fails, Flyte invokes clean_up, passing the original name input and the error that caused the failure. The FlyteError object provides the message and the failed_node_id to help diagnose which specific step in the DAG triggered the handler.