Workflow composition, failure handlers, and nodes

Flytekit workflows are defined by decorating Python functions with @workflow. While these functions look like standard Python, they serve a dual purpose: they are executed locally for testing and compiled into a Directed Acyclic Graph (DAG) of Node objects for remote execution on Flyte.

Workflow Composition and Dataflow

In flytekit, workflow composition is achieved by passing the outputs of one task or subworkflow as inputs to another. During compilation, these outputs are represented by Promise objects.

Task Outputs and Promises

When you call a task inside a workflow, it returns a Promise (or a collection of Promises). A Promise is a placeholder for a value that will be computed at runtime.

@workflow
def my_wf(a: int) -> int:
    # x is a Promise object during compilation
    x = add_5(a=a)
    # Passing the promise x to another task creates a dependency edge
    return add_5(a=x)

If a task returns multiple values, flytekit wraps them in a NamedTuple of Promises. You can access individual outputs using attribute notation or indexing:

@workflow
def multi_output_wf(a: int) -> int:
    # Assume t1 returns (int, str)
    o = t1(a=a)
    # Access by attribute or index
    t2(x=o.o0) 
    t3(y=o[1])
    return o.o0

Manual Node Creation

While standard task calls are sufficient for most dataflows, create_node allows for explicit control over node construction. This is useful for tasks with side effects where no data is returned, or when you need to manually manage dependencies.

Unlike ordinary task calls that return Promises, create_node returns a Node object. You must access its outputs via the .outputs attribute or shorthand properties like .o0.

from flytekit import workflow, create_node

@workflow
def manual_node_wf(a: int):
    # create_node returns a Node object, not a Promise
    t1_node = create_node(t1, a=a)
    t2_node = create_node(t2)
    
    # Use >> to define execution order without dataflow
    t1_node >> t2_node
    
    # Access outputs from the Node object
    t3(x=t1_node.o0)
    t4(y=t1_node.outputs["o1"])

Note: The .outputs property is only available on Node objects created via create_node. Calling .outputs on a standard Node (internal to flytekit) will raise an AssertionError.

Per-Node Overrides

You can customize the execution behavior of specific nodes using .with_overrides(). This method is available on both Promise objects and Node objects. Overrides are static and cannot be determined by other task outputs (Promises).

from flytekit import Resources, workflow

@workflow
def override_wf(a: int) -> int:
    return add_5(a=a).with_overrides(
        requests=Resources(cpu="500m", mem="1Gi"),
        retries=3,
        interruptible=True,
        node_name="custom-add-node"
    )

Supported Overrides

The Node.with_overrides method supports several parameters:

• Resources: Set requests and limits using flytekit.Resources. Note that requests are clamped to original limits if limits are not also overridden.
• Retries: An integer specifying the number of retry attempts.
• Timeout: A datetime.timedelta or integer seconds.
• Cache: A boolean or Cache object. If using a Cache object, you must specify a version.
• Metadata: interruptible status and custom node_name.

Failure Handlers

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

Defining a Failure Handler

A valid failure handler must:

1. Accept all inputs defined in the parent workflow.
2. Optionally accept an additional parameter for the error, which must be typed as typing.Optional[FlyteError].

from typing import Optional
from flytekit import workflow, task, FlyteError

@task
def clean_up(name: str, err: Optional[FlyteError] = None):
    if err:
        print(f"Workflow failed at node {err.failed_node_id} with error: {err.message}")
    print(f"Cleaning up resources for {name}")

@workflow(on_failure=clean_up)
def my_wf(name: str):
    t1(name=name)
    t2()

Internal Behavior

When a failure occurs:

1. Flytekit catches the exception.
2. It injects a FlyteError object into the handler's err (or error) input. This object contains the message and the failed_node_id.
3. The handler is executed.
4. After the handler completes, the original exception is re-raised to ensure the workflow is marked as failed.

Imperative Workflows

For dynamic DAG construction, ImperativeWorkflow provides a programmatic interface to add inputs, nodes, and outputs.

from flytekit.core.workflow import ImperativeWorkflow

wb = ImperativeWorkflow(name="my_imperative_wf")
wb.add_workflow_input("in1", int)

# add_entity returns a Node
node = wb.add_entity(add_5, a=wb.inputs["in1"])

# Bind workflow output to a node output
wb.add_workflow_output("out1", node.outputs["o0"])

# Add a failure handler
wb.add_on_failure_handler(clean_up)

In imperative workflows, add_entity internally uses create_node. The resulting Node objects are used to wire dependencies and define workflow outputs via node.outputs.

Common Pitfalls

• Truth Value Testing: You cannot use if promise: or bool(promise) inside a workflow function because Promises are not resolved during compilation. Use the conditional construct instead.
• Logical Operators: Use & and | for logical operations on Promises. Standard Python and and or will raise a ValueError.
• Iteration: You cannot iterate over a Promise (e.g., for i in promise_list:). Workflows must use map_task or dynamic tasks for variable-length iteration.
• VoidPromise: Tasks that return None return a VoidPromise. These cannot be used as inputs to other tasks but can be used for sequencing with >>.
• Positional Arguments: create_node only accepts keyword arguments for task inputs. Passing positional arguments will result in a FlyteAssertion.