Workflow composition, failure handlers, and nodes

In flytekit, workflows are composed of nodes that represent units of execution. While the standard @workflow decorator allows for intuitive Pythonic composition, the underlying engine uses Node objects and Promise references to manage data flow and execution order.

Workflow Composition and Promises

When you define a workflow, flytekit transforms the decorated function into a PythonFunctionWorkflow. Inside this function, calling a task does not return the actual value (like an int or str). Instead, it returns a Promise object.

A Promise acts as a placeholder for a future value. It allows flytekit to build a directed acyclic graph (DAG) of tasks during compilation.

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, not an int
    val = get_value()
    # The Promise is passed to the next task to create a dependency
    return process_value(v=val)

Internally, the Promise class (found in flytekit.core.promise) tracks whether it is "ready" (has a value during local execution) or is a reference to a NodeOutput. You cannot use a Promise in standard Python control flow like if my_promise:, as this will raise a ValueError. Instead, use comparison operators like == or !=, which return a ComparisonExpression for flytekit's internal engine.

Explicit Node Creation

While standard task calls are sufficient for most cases, create_node (from flytekit.core.node_creation) provides explicit control over node instantiation. This is particularly useful for:

1. Ordering tasks without data dependencies: Using the >> operator or runs_before method.
2. Accessing outputs by name: Useful in imperative workflow patterns.
3. Applying per-node overrides: Configuring resources or retries for a specific instance of a task.

Accessing Node Outputs

When you use create_node, the outputs are not returned as a single Promise or tuple. Instead, they are attached to the returned Node object. You can access them as attributes (e.g., .o0, .o1) or via the .outputs dictionary.

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

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

@task
def consumer(i: int):
    print(i)

@workflow
def explicit_wf():
    node = create_node(multi_output)
    
    # Accessing via attribute
    consumer(i=node.o0)
    
    # Accessing via .outputs dictionary
    consumer(i=node.outputs["o0"])

Note that node.outputs is only available for nodes created via create_node. Standard Node objects used internally by flytekit will raise an AssertionError if you attempt to access .outputs directly.

Per-Node Overrides

The Node class provides a with_overrides method to customize execution parameters for a specific step. This can be called on the result of create_node or on a Promise returned by a task call.

@workflow
def override_wf(val: int):
    # Overriding on a Promise
    t1_promise = task_one(a=val).with_overrides(node_name="custom-t1", retries=3)
    
    # Overriding on an explicit Node
    t2_node = create_node(task_two, b=t1_promise)
    t2_node.with_overrides(timeout=3600, interruptible=True)

The with_overrides method (implemented in flytekit.core.node) updates the NodeMetadata, including retries, timeout, and interruptible status. It also handles resource requests and limits via the Resources class.

Failure Handlers

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

A valid failure handler must follow specific signature rules:

1. It must accept the same inputs as the workflow (or a subset).
2. It can optionally accept an err parameter of type FlyteError to inspect the failure.
3. Any additional parameters must be Optional.

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

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

@task
def failing_task(name: str):
    raise ValueError(f"Failing execution for {name}")

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

When a node in failure_wf fails, flytekit catches the exception and triggers the clean_up task, passing the original workflow inputs and the error details. This ensures that resources can be released or alerts can be sent even when the primary logic fails.