Workflow composition, failure handlers, and nodes

Flytekit workflows are defined using the @workflow decorator, which transforms a Python function into a directed acyclic graph (DAG) of nodes. During compilation, task calls within the workflow body do not return native Python values; instead, they return Promise objects that represent future results.

Workflow Composition and Promises

When you call a task inside a workflow, flytekit records the call as a Node in the workflow graph. The return value is a Promise (or a tuple of Promises) that points to that node's output.

import typing
from flytekit import task, workflow

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

@task
def multiply(x: int, y: int) -> int:
    return x * y

@workflow
def math_workflow(a: int, b: int) -> int:
    # x is a Promise, not an int
    x = add_one(x=a)
    # You can pass Promises directly into other tasks
    result = multiply(x=x, y=b)
    return result

Accessing Structured Outputs

If a task returns multiple values (e.g., via a NamedTuple), the Promise allows you to access specific attributes. Internally, Promise.__getattr__ and Promise.__getitem__ append the attribute name or index to an attr_path, which is resolved when the workflow executes.

class MyOutputs(typing.NamedTuple):
    res: int
    msg: str

@task
def complex_task(x: int) -> MyOutputs:
    return MyOutputs(res=x * 2, msg="Success")

@workflow
def attribute_workflow(a: int) -> int:
    o = complex_task(x=a)
    # Accessing .res creates a new Promise with an updated attribute path
    return o.res

Promise Limitations

Because Promise objects are placeholders for future values, they cannot be used in standard Python control flow or operations:

• No Truth Testing: if my_promise: will raise a ValueError.
• No Iteration: for i in my_promise: or range(my_promise) will fail.
• Logical Operators: Use bitwise & and | for logical AND/OR in conditionals, as Python's and/or cannot be overridden to return a graph expression.

Explicit Node Creation

While ordinary task calls are the standard way to build workflows, create_node provides lower-level control. This is useful for tasks that have no outputs but must run in a specific order, or when you need to access node-level metadata.

A critical distinction in flytekit is that ordinary task calls do not return a Node object. They return a Promise. If you need to access the outputs dictionary of a node, you must use create_node.

from flytekit import task, workflow, create_node

@task
def t1(a: int) -> int:
    return a + 1

@workflow
def manual_node_wf(a: int) -> int:
    # create_node returns a Node object, not a Promise
    node = create_node(t1, a=a)
    
    # Access outputs via .o0 (default first output) or the .outputs dict
    # This is only possible with create_node
    return node.outputs["o0"]

Dependency Chaining

You can enforce execution order between nodes that do not share data using the >> operator or the runs_before method. This is implemented in node.Node.runs_before, which appends the current node to the upstream list of another node.

@task
def side_effect_task():
    print("Side effect")

@workflow
def dependency_wf():
    n1 = create_node(side_effect_task)
    n2 = create_node(side_effect_task)
    
    # n1 will execute before n2
    n1 >> n2

Per-Node Overrides

You can override task-level configurations (like resource limits, retries, or timeouts) on a per-node basis within a workflow. These overrides can be applied to either a Promise or a Node.

When called on a Promise, the with_overrides method forwards the configuration to the underlying Node via self.ref.node.with_overrides.

from flytekit import Resources

@workflow
def override_wf(a: int) -> int:
    # Applying overrides to a Promise
    promise_res = add_one(x=a).with_overrides(
        requests=Resources(cpu="500m", mem="1Gi"),
        retries=3,
        node_name="custom-add-node"
    )
    
    # Applying overrides to a Node
    node = create_node(multiply, x=promise_res, y=a)
    node.with_overrides(timeout=3600)
    
    return node.o0

Note: Overrides like node_name, retries, and timeout must be static values. Passing a Promise into with_overrides for these parameters will raise an error during compilation.

Workflow Failure Handlers

The @workflow decorator accepts an on_failure parameter to specify a task or subworkflow that should run if any node in the workflow fails.

Signature Requirements

The failure handler must follow strict signature rules validated by PythonFunctionWorkflow._validate_add_on_failure_handler:

1. It must accept every input defined in the main workflow.
2. It may accept an additional optional parameter named err (of type flytekit.models.core.errors.FlyteError).
3. Any other parameters in the handler must be Optional and have default values.

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

@task
def cleanup_task(a: int, err: Optional[FlyteError] = None):
    if err:
        print(f"Workflow failed on node {err.failed_node_id} with: {err.message}")
    print(f"Cleaning up resources for input {a}")

@workflow(on_failure=cleanup_task)
def failure_handling_wf(a: int) -> int:
    return add_one(x=a)

Internally, flytekit compiles the on_failure entity into a special failure node. When a failure occurs, Flyte invokes this node with the original workflow inputs and injects the FlyteError into the err parameter. After the handler completes, the original exception is re-raised.

Imperative Workflows

For dynamic scenarios where the workflow structure isn't known until runtime, flytekit provides the ImperativeWorkflow class. This allows you to manually add inputs, entities (tasks/workflows), and outputs.

from flytekit import ImperativeWorkflow

wb = ImperativeWorkflow(name="dynamic_math")
# Add workflow inputs
in_a = wb.add_workflow_input("a", int)

# Add a task as a node
node = wb.add_entity(add_one, x=in_a)

# Add workflow outputs using the node's output dictionary
wb.add_workflow_output("result", node.outputs["o0"])

# Add a failure handler
wb.add_on_failure_handler(cleanup_task)

The add_entity method internally uses create_node to link the task into the graph and returns the Node object, allowing you to access node.outputs for subsequent bindings.