Workflow composition, failure handlers, and nodes

When you define a workflow in flytekit using the @workflow decorator, the function body is executed at compile-time to build a Directed Acyclic Graph (DAG). Because this happens before any actual data exists, tasks do not return real values; instead, they return Promise objects that act as placeholders for future results.

Workflow Composition and Promises

In flytekit, you compose workflows by passing the output of one task as the input to another. Internally, the Promise class (found in flytekit.core.promise) manages these data dependencies.

from flytekit import task, workflow

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

@task
def shout(text: str) -> str:
    return f"{text.upper()}!!!"

@workflow
def greeting_workflow(name: str) -> str:
    # greeting is a Promise object, not a string
    greeting = get_greeting(name=name)
    # Passing a Promise to another task links the nodes in the DAG
    return shout(text=greeting)

Because greeting is a Promise, you cannot perform standard Python operations on it inside the @workflow function. For example, len(greeting) or if greeting == "Hello": will fail during compilation because the actual value is unknown.

Accessing Attributes and Indexing

If a task returns a complex type like a dataclass or a dict, you can access its attributes or keys using standard Python syntax. The Promise.__getattr__ and Promise.__getitem__ methods intercept these calls and return a new Promise with an updated attr_path.

@task
def get_data() -> dict:
    return {"key": "value", "list": [1, 2, 3]}

@workflow
def attribute_wf() -> int:
    data = get_data()
    # Returns a Promise pointing to the specific index/key
    return data["list"][0] 

Explicit Node Creation

While calling a task directly is the most common way to add a node to a workflow, flytekit.core.node_creation.create_node provides lower-level control. This is useful for:

1. Establishing execution order without data dependencies.
2. Accessing outputs by name when programmatically building workflows.

Execution Dependencies

You can use the >> operator or the runs_before method on Node objects to ensure one task finishes before another starts, even if they don't share data.

from flytekit.core.node_creation import create_node

@workflow
def ordered_wf():
    n1 = create_node(task_a)
    n2 = create_node(task_b)
    
    # task_a will always run before task_b
    n1 >> n2 

Accessing Node Outputs

When you use create_node, the return value is a Node object (or a VoidPromise if the task has no outputs). The outputs of the node are accessible via the .outputs dictionary or as attributes named o0, o1, etc.

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

@workflow
def node_output_wf():
    node = create_node(multi_output)
    
    # Accessing outputs from the Node object
    use_task(val=node.o0)
    use_task(val=node.outputs["o1"])

Note that create_node(...).outputs is only available on Node objects. Ordinary task calls return Promise objects, which do not have an .outputs attribute.

Per-Node Overrides

The Node.with_overrides method allows you to customize the execution environment for a specific task instance within a workflow. You can override resources, retries, timeouts, and more.

from flytekit import Resources

@workflow
def override_wf(val: int):
    # Override resources and retries for this specific task call
    task_a(val=val).with_overrides(
        requests=Resources(cpu="2", mem="4Gi"),
        retries=3,
        node_name="high-priority-task"
    )

Internally, with_overrides modifies the NodeMetadata and Resources associated with the Node. If called on a Promise, it delegates the call to the underlying Node via self.ref.node.

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.

Signature Requirements

A valid on_failure handler must:

1. Accept all inputs defined in the main workflow.
2. Optionally accept an err parameter of type flytekit.models.core.errors.FlyteError.

from typing import Optional
from flytekit.models.core.errors import FlyteError

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

@workflow(on_failure=cleanup_task)
def main_wf(name: str):
    # If any task here fails, cleanup_task is invoked with 'name' 
    # and the resulting FlyteError.
    risky_task(name=name)

The on_failure handler is treated as a special node in the Flyte DAG that triggers only upon the failure of any other node in the workflow. It ensures that resources can be released or stakeholders notified regardless of where the failure occurred.