Workflow composition, failure handlers, and nodes

Workflows in flytekit are composed of tasks, sub-workflows, and other entities organized into a directed acyclic graph (DAG). While most workflows are defined by passing data between tasks, flytekit provides lower-level primitives for explicit node management, failure handling, and per-node configuration.

Workflow Composition and Promises

When you call a task inside a function decorated with @workflow, it does not execute immediately. Instead, it returns one or more Promise objects. These objects represent future values that will be available when the task completes during execution.

from flytekit import task, workflow

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

@task
def greet(greeting: str):
    print(greeting)

@workflow
def welcome_wf(name: str):
    # get_greeting returns a Promise[str]
    greeting_promise = get_greeting(name=name)
    
    # Passing the promise to another task creates a data dependency
    greet(greeting=greeting_promise)

Internally, the Promise class (found in flytekit/core/promise.py) acts as a wrapper. During compilation, it holds a NodeOutput reference which points to the origin Node that produces the value.

Logical Operations on Promises

Because Promise objects are not actual values during workflow definition, you cannot use standard Python logical operators like if promise: or and/or. flytekit overrides bitwise operators and provides specific methods for these expressions:

• Equality/Comparison: Use ==, !=, <, >, <=, >=. These return a ComparisonExpression.
• Logical Conjunctions: Use & (AND) and | (OR). These return a ConjunctionExpression.
• Boolean Methods: Use .is_true(), .is_false(), or .is_none().

# Inside a workflow or conditional block
result = my_task()
# Correct:
(result == "success") & (other_task().is_true())
# Incorrect:
if result == "success": ...

Explicit Node Creation

While calling tasks directly is the standard way to build workflows, create_node (in flytekit/core/node_creation.py) allows for explicit control over the underlying Node objects. This is useful for:

1. Non-data dependencies: Forcing one task to run after another when no data is shared.
2. Imperative style: Building workflows where output names are determined dynamically.

Dependency Management

You can use the >> operator or the runs_before method to define execution order between nodes.

from flytekit import task, workflow
from flytekit.core.node_creation import create_node

@task
def setup(): ...

@task
def work(): ...

@workflow
def manual_wf():
    setup_node = create_node(setup)
    work_node = create_node(work)
    
    # setup must finish before work starts
    setup_node >> work_node 

Accessing Outputs from create_node

A critical distinction exists between calling a task and using create_node. A task call returns a Promise (or a tuple of them). create_node returns a Node object. To access the outputs of a Node, you must use the attribute naming convention o0, o1, etc., or the .outputs dictionary.

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

@workflow
def output_wf():
    node = create_node(compute)
    
    # Accessing outputs by attribute (o0, o1, ...)
    other_task(val=node.o0)
    
    # Accessing via the outputs dictionary
    another_task(val=node.outputs["o1"])

Failure Handlers

Workflows can define an on_failure handler to perform cleanup or notification when a workflow execution fails. The handler is typically a task or another workflow.

The failure handler must follow specific signature rules:

1. It must accept all inputs that the original workflow accepts.
2. It can optionally accept an additional argument of type FlyteError (usually named err).

import typing
from flytekit import task, workflow
from flytekit.models.core.errors 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"Cleaning up for {name}")

@workflow(on_failure=clean_up)
def wf(name: str):
    # If any task here fails, clean_up(name=name, err=...) is invoked
    t1(a=1)

Per-Node Overrides

You can customize the behavior of individual nodes (tasks or sub-workflows) using the with_overrides method. This method is available on both Promise objects and Node objects.

Common overrides include:

• requests and limits: Resource requirements (CPU, Memory).
• timeout: Maximum execution time.
• retries: Number of times to retry on failure.
• interruptible: Whether the node can run on spot/preemptible instances.

from flytekit import Resources

@workflow
def override_wf(val: int):
    # Overriding on a Promise (returned by task call)
    promise = my_task(val=val).with_overrides(
        retries=3,
        requests=Resources(cpu="2", mem="500Mi")
    )
    
    # Overriding on a Node (returned by create_node)
    node = create_node(another_task, val=promise)
    node.with_overrides(node_name="custom-node-name", timeout=3600)

The Node.with_overrides implementation (in flytekit/core/node.py) updates the NodeMetadata and resource specifications. Note that if you provide resources, you cannot also provide limits or requests in the same call, as resources is a shorthand for both.