Workflow composition, failure handlers, and nodes

Flytekit workflows are constructed by connecting tasks, sub-workflows, and launch plans into a Directed Acyclic Graph (DAG). This composition relies on Promise objects to track data flow and Node objects to represent execution steps.

Workflow Composition and Promises

When you call a task inside a @workflow function, it does not return the actual value. Instead, it returns a Promise (defined in flytekit.core.promise.Promise). This object acts as a placeholder for a future value that will be produced during execution.

from flytekit import task, workflow

@task
def get_greeting(name: str) -> str:
    return f"Hello, {name}!"

@task
def greet(greeting: str):
    print(greeting)

@workflow
def welcome_wf(name: str):
    # 'greeting' is a Promise[str], not a string
    greeting = get_greeting(name=name)
    greet(greeting=greeting)

Internally, a Promise wraps a NodeOutput, which contains the ID of the node that produces the value and the name of the output variable. During local execution, the Promise is "ready" and holds a literal value in its val attribute. During compilation (registration), it is "not ready" and uses its ref attribute to point to the upstream node.

Logical Operations on Promises

Because Promise objects are not actual values during workflow construction, you cannot use standard Python logical operators like if greeting: or and/or. Flytekit provides bitwise operator overrides and helper methods for logical expressions:

• Use & for AND and | for OR.
• Use .is_true(), .is_false(), or .is_none() for boolean comparisons.

These operations return a ComparisonExpression or ConjunctionExpression (from flytekit.core.promise), which Flyte uses to build conditional branches.

Explicit Node Creation

While most nodes are created implicitly by calling tasks, you can use create_node from flytekit.core.node_creation to explicitly manage a node. This is useful when you need to establish execution order without a direct data dependency.

Establishing Dependencies

You can use the >> operator (or the runs_before method) to force one node to run after another.

from flytekit.core.node_creation import create_node

@workflow
def dependency_wf():
    node_a = create_node(task_a)
    node_b = create_node(task_b)
    
    # task_b will only run after task_a completes successfully
    node_a >> node_b

Accessing Outputs from create_node

A critical distinction between a direct task call and create_node is how you access outputs. A task call returns a Promise (or a tuple of them). create_node returns a Node object, and its outputs are attached as attributes named o0, o1, etc., or accessible via the outputs dictionary.

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

@workflow
def output_wf():
    # Direct call returns a tuple of Promises
    p1, p2 = multi_output()
    
    # create_node returns a Node object
    node = create_node(multi_output)
    # Access outputs via attributes or the .outputs property
    val_int = node.o0
    val_str = node.outputs["o1"]

Per-Node Overrides

You can customize the execution parameters of a specific node using the with_overrides method. This is available on both Promise objects (returned by task calls) and Node objects (returned by create_node).

Common overrides include:

• requests and limits: Resource requirements using flytekit.Resources.
• timeout: A datetime.timedelta or integer seconds.
• retries: Number of times to retry on failure.
• node_name: A custom name for the node in the Flyte UI.

from flytekit import Resources

@workflow
def override_wf(n: int):
    # Overriding resources on a task call
    t1 = task_a(n=n).with_overrides(
        requests=Resources(cpu="2", mem="500Mi"),
        node_name="heavy-computation"
    )
    
    # Overriding on a node object
    node_b = create_node(task_b).with_overrides(retries=3)

The Node.with_overrides method (in flytekit/core/node.py) validates these inputs and updates the NodeMetadata or Resources model associated with that specific step in the DAG.

Failure Handlers

Workflows can define an on_failure handler using the @workflow decorator. This handler is triggered if any node in the workflow fails.

Signature Requirements

The failure handler must be a task or workflow that:

1. Accepts all inputs of the parent workflow.
2. Can optionally accept an error parameter of type FlyteError to inspect the failure details.

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

@task
def cleanup_task(name: str, error: FlyteError):
    print(f"Workflow failed for {name} with error: {error.message}")

@workflow(on_failure=cleanup_task)
def main_wf(name: str):
    # If this task fails, cleanup_task is invoked with 'name' and the error
    risky_task(name=name)

When a failure occurs, Flytekit ensures the on_failure entity is executed with the original workflow inputs. If your handler requires additional inputs not present in the workflow signature, they must be defined as Optional or have default values.