Workflow composition, failure handlers, and nodes

Flytekit workflows are declarative structures that define a Directed Acyclic Graph (DAG) of tasks. While the workflow body looks like standard Python, it is evaluated at compile-time to build this graph, using Promise objects to represent data flow and Node objects to represent execution steps.

Workflow Composition and Promises

When you call a task inside a @workflow decorated function, it does not return the actual value (like an int or str). Instead, it returns a flytekit.core.promise.Promise. This promise is a placeholder for a value that will be produced during execution.

from flytekit import task, workflow

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

@workflow
def greeting_workflow(name: str) -> str:
    # 'greeting' is a Promise, not a string
    greeting = get_greeting(name=name)
    return greeting

Because these are promises, you cannot perform standard Python operations on them inside the workflow body, such as len(promise) or if promise:. Attempting to iterate over a promise will raise a ValueError from the Promise.__iter__ method in flytekit/core/promise.py.

Accessing Task Outputs

If a task returns multiple values, Flytekit returns a Promise that can be indexed or accessed via attributes, depending on how the task's return type is defined (e.g., using typing.NamedTuple).

Internally, Promise.__getattr__ and Promise.__getitem__ call _append_attr, which tracks the attribute path. This path is used during execution to resolve the specific output from the node.

Explicit Node Creation

In most cases, Flytekit creates nodes automatically when you call a task. However, you can use flytekit.core.node_creation.create_node to explicitly create a Node object. This is useful for:

1. Ordering dependencies between tasks that do not share data.
2. Applying overrides to a specific execution unit.

Ordering Dependencies

If task_b must run after task_a, but task_b does not take any inputs from task_a, you can use the >> operator or the runs_before method.

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

@task
def setup():
    ...

@task
def compute():
    ...

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

Accessing Outputs from create_node

Unlike a direct task call which returns a Promise, create_node returns a Node object. To access the outputs of the task within that node, you must use the .o0, .o1, etc., attributes (or named attributes if using a NamedTuple).

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

@task
def consume_data(val: int):
    ...

@workflow
def manual_node_wf():
    producer = create_node(produce_data)
    # Access the first output via .o0
    consume_data(val=producer.o0)

Per-Node Overrides

You can customize the execution parameters of a specific node using the with_overrides method. This method is available on both Promise objects (where it applies to the upstream node) and Node objects.

Common overrides include:

• requests and limits: Resource requirements using flytekit.Resources.
• retries: Number of times to retry a failed node.
• timeout: Maximum execution time as an int (seconds) or datetime.timedelta.
• node_name: A custom name for the node in the Flyte UI.

from flytekit import Resources

@workflow
def resource_wf(val: int):
    # Overriding resources on a task call promise
    t1 = get_greeting(name="flyte").with_overrides(
        requests=Resources(cpu="1", mem="2Gi"),
        retries=3,
        node_name="custom-greeting-node"
    )

The Node.with_overrides implementation in flytekit/core/node.py validates these inputs, ensuring that resource requests are not Promise objects and that node_name is DNS-compliant via the _dnsify utility.

Failure Handlers

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

Requirements for on_failure Handlers

A valid on_failure handler must:

1. Accept an err argument of type typing.Optional[flytekit.types.error.error.FlyteError].
2. Accept all inputs that the workflow itself accepts.
3. Any additional arguments must be Optional.

from typing import Optional
from flytekit import task, workflow
from flytekit.types.error.error import FlyteError

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

@workflow(on_failure=clean_up)
def my_wf(name: str):
    # If any task here fails, clean_up(name=name, err=...) is invoked
    ...

When a workflow fails, Flytekit captures the error and passes it to the on_failure entity. The err object contains the message and the failed_node_id that triggered the failure. In local execution, this behavior is simulated by catching exceptions and invoking the handler before re-raising the error.