Workflow composition, failure handlers, and nodes

Flytekit uses a declarative model for workflow composition. When you decorate a function with @workflow, flytekit evaluates the function body to construct a Directed Acyclic Graph (DAG) of nodes and data-flow bindings. During this compilation phase, task calls do not return actual Python values; instead, they return symbolic Promise objects that represent future values.

Workflow Composition and Promises

In a flytekit workflow, data flows between tasks via Promise objects. A Promise acts as a proxy for a value that will be produced at runtime.

Task Outputs and Data Flow

When you call a task within a workflow, it returns a Promise (for a single output), a tuple of Promise objects, or a VoidPromise (for tasks with no outputs). These promises can be passed directly into other tasks.

from flytekit import task, workflow
import typing

@task
def t1(a: int) -> typing.NamedTuple("Outputs", [("val", int), ("msg", str)]):
    return a + 2, f"result-{a}"

@task
def t2(val: int) -> int:
    return val * 2

@workflow
def my_wf(a: int) -> int:
    # x and y are Promise objects during compilation
    x, y = t1(a=a)
    # Passing a promise into another task
    result = t2(val=x)
    return result

Symbolic Operations on Promises

Because promises are symbolic, you cannot perform standard Python operations like range(promise) or truth testing (if promise:) inside the workflow body. However, flytekit supports symbolic attribute and index access which are resolved at runtime.

@workflow
def structured_wf():
    o = t1(a=10)
    # Appends attribute path to the promise for later resolution
    t2(val=o.val) 
    # Indexing is also supported for lists/dicts
    # t3(x=o["some_key"][0])

Internally, Promise.__getattr__ and Promise.__getitem__ in flytekit/core/promise.py create a copy of the promise and append the key to its attr_path.

Explicit Node Creation

While standard task calls return promises, you can use create_node from flytekit.core.node_creation to manually construct a Node. This is useful when you need to define execution dependencies without data dependencies.

create_node vs. Task Calls

A standard task call returns a Promise or VoidPromise. In contrast, create_node returns a Node object during compilation.

• Task Call: Returns Promise. Used for data flow.
• create_node: Returns Node. Used for graph manipulation and explicit ordering.

from flytekit.core.node_creation import create_node

@workflow
def explicit_wf(a: int):
    # create_node requires keyword arguments
    n1 = create_node(t1, a=a)
    n2 = create_node(t2, val=10)

    # Explicit ordering using shift operator or runs_before
    n1 >> n2 
    # n1.runs_before(n2)

Accessing Outputs from Nodes

Unlike ordinary nodes, nodes created via create_node expose their outputs through a .outputs dictionary and as attributes (e.g., .o0, .o1).

@workflow
def output_access_wf(a: int):
    n1 = create_node(t1, a=a)
    # Accessing the first output promise from the node
    t2(val=n1.outputs["val"])
    # Or via attribute
    t2(val=n1.val)

Per-Node Overrides

You can override execution parameters for specific nodes using the .with_overrides() method. This method is available on both Node objects and Promise objects (where it is forwarded to the underlying node).

Supported Overrides

Overrides allow you to modify metadata and resource requirements defined in flytekit/core/node.py:

• Resources: requests, limits, accelerator, shared_memory.
• Metadata: node_name, timeout, retries, interruptible, cache.
• Container: container_image, pod_template.

from flytekit import Resources

@workflow
def override_wf(a: int) -> int:
    promise = t2(val=a).with_overrides(
        node_name="my-custom-node",
        requests=Resources(cpu="2", mem="500Mi"),
        retries=3,
        timeout=3600
    )
    return promise

Note: Overrides like retries, cache, and resources must be static values. Passing a Promise to these arguments will raise an error during compilation.

Failure Handlers

The @workflow decorator supports an on_failure parameter to specify a task or sub-workflow that runs if the main workflow fails.

Signature Requirements

The failure handler must follow strict signature rules validated in flytekit/core/workflow.py:

1. It must accept every input defined in the workflow's signature.
2. Any additional inputs in the handler must be Optional.
3. An optional err parameter of type flytekit.FlyteError can be included to receive details about the failure.

from flytekit import FlyteError
import typing

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

@workflow(on_failure=clean_up)
def wf_with_handler(a: int) -> int:
    return t2(val=a)

Execution Behavior

During local execution, if a workflow fails, flytekit catches the exception, invokes the on_failure handler with the provided inputs and the FlyteError (if declared), and then re-raises the original exception. In a remote environment, the Flyte engine manages the execution of the failure node as a separate step in the DAG.