Workflow composition, failure handlers, and nodes

Workflows in flytekit are declarative entities that construct a Directed Acyclic Graph (DAG) of tasks. Unlike tasks, which are evaluated at runtime, the body of a @workflow function is evaluated at serialization time to build the execution graph.

Workflow Composition and Promises

When you call a task inside a workflow, flytekit does not immediately execute the task. Instead, it creates a Node in the DAG and returns Promise objects. A Promise (defined in flytekit.core.promise.Promise) acts as a placeholder for a future value.

from flytekit import task, workflow

@task
def add_one(x: int) -> int:
    return x + 1

@workflow
def my_workflow(val: int) -> int:
    # result is a Promise, not an int
    result = add_one(x=val)
    return result

Internally, the Promise class handles the duality between compilation and local execution. During compilation, it holds a NodeOutput reference to the upstream node. During local execution, it stores the actual computed Literal value.

Accessing Task Outputs

If a task returns multiple values (e.g., a tuple or NamedTuple), the task call returns a collection of promises. You can access individual outputs using standard Python indexing or attribute access.

import typing
from flytekit import task, workflow

@task
def multi_output() -> typing.NamedTuple("Outputs", [("out_int", int), ("out_str", str)]):
    return 1, "hello"

@workflow
def wf() -> (int, str):
    res = multi_output()
    # Accessing by attribute or index returns a new Promise with an updated attr_path
    return res.out_int, res[1]

The Promise.__getitem__ and Promise.__getattr__ methods in flytekit.core.promise append the key to the attr_path, ensuring that the correct output is resolved during execution.

Explicit Node Creation

While calling tasks directly is the standard way to compose workflows, flytekit.core.node_creation.create_node provides lower-level control. This is useful for:

1. Non-data dependencies: Forcing a task to run after another even if they don't share data.
2. Accessing Node attributes: Accessing the underlying Node object to apply overrides or inspect metadata.

from flytekit import task, workflow, create_node

@task
def t1(): ...

@task
def t2(): ...

@workflow
def wf():
    node_1 = create_node(t1)
    node_2 = create_node(t2)
    
    # Explicitly set execution order using the shift operator
    node_1 >> node_2 

Distinguishing Outputs

A critical distinction exists between standard task calls and create_node:

• Standard Task Call: Returns a Promise (or a tuple of them). You cannot access .outputs on these.
• create_node: Returns a Node object. You access its outputs via the .outputs attribute or as direct attributes on the node (e.g., node.o0).

As defined in flytekit.core.node.Node.outputs, calling .outputs on a node not created via create_node() will raise an AssertionError.

Per-Node Overrides

You can customize the execution parameters of a specific task instance within a workflow using the with_overrides method. This method is available on both Node objects and Promise objects (which forward the call to their underlying node).

from flytekit import task, workflow, Resources

@task
def compute_task(x: int) -> int: ...

@workflow
def wf(x: int) -> int:
    return compute_task(x=x).with_overrides(
        requests=Resources(cpu="2", mem="4Gi"),
        retries=3,
        node_name="special-compute-node"
    )

The Node.with_overrides method in flytekit.core.node supports several parameters:

• requests and limits: Resource specifications using flytekit.Resources.
• timeout: A datetime.timedelta or integer seconds.
• retries: Number of retry attempts.
• interruptible: Boolean indicating if the node can be run on spot/preemptible instances.
• cache and cache_version: Overriding caching behavior.

Note: You cannot use Promise objects for resource overrides; these must be concrete values.

Failure Handlers

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

A valid failure handler must:

1. Accept all inputs that the workflow itself accepts.
2. Optionally accept an err argument of type typing.Optional[FlyteError].

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

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

@workflow(on_failure=clean_up)
def my_wf(name: str):
    # ... workflow logic ...
    pass

When a failure occurs, flytekit captures the exception, wraps it in a FlyteError (containing the failed_node_id and message), and invokes the on_failure entity with the original workflow inputs and the error object. This logic is implemented in WorkflowBase.__call__ within flytekit/core/workflow.py.