Workflow composition, failure handlers, and nodes

When authoring workflows in flytekit, tasks do not execute immediately to produce concrete Python values. Instead, flytekit compiles workflow functions into a declarative directed acyclic graph (DAG) composed of execution nodes and future data bindings. Understanding how workflow, create_node, Promise, and node overrides interact prevents runtime surprises—such as attempting to iterate over promises or encountering mismatched signatures in workflow failure handlers.

Workflow Composition and Promise Dataflow

Flytekit workflows are declared using the @workflow decorator, which turns a Python function into an instance of PythonFunctionWorkflow (inheriting from WorkflowBase).

from flytekit import task, workflow


@task
def generate_seed() -> int:
    return 42


@task
def compute(seed: int) -> str:
    return f"result-{seed * 2}"


@workflow
def my_workflow() -> str:
    seed_promise = generate_seed()
    result_promise = compute(seed=seed_promise)
    return result_promise

The Compilation vs. Execution Duality

When flytekit evaluates @workflow definitions during workflow registration or compilation, calling a task function like generate_seed() does not execute its Python body. Instead:

1. FlyteContextManager.current_context() contains an active CompilationState.
2. The entity invocation triggers flytekit's internal flyte_entity_call_handler and create_and_link_node, constructing a Node instance in ctx.compilation_state.nodes.
3. The call returns a Promise instance (or a custom namedtuple of Promise instances for multiple return values, or VoidPromise if the task returns None).

A Promise contains a reference ref: NodeOutput, tracking the upstream node ID, output variable name, and optional nested attribute path (attr_path).

During local execution (e.g., calling my_workflow() directly in pytest or a Python script), flytekit executes the underlying task functions in ExecutionState.Mode.LOCAL_WORKFLOW_EXECUTION, returning actual Python values wrapped inside resolved Promise instances whose is_ready property is True.

Expression and Iteration Constraints on Promises

Because a Promise represents a graph edge rather than a computed value at compile time, Python truth-value testing and standard control flow fail:

• Truth testing and logic operators: Calling bool(promise), if promise:, promise and other, or promise or other raises a ValueError. To build conditions in Flytekit, use bitwise operators & and | with comparison expressions (such as promise == 10 or promise.is_true()), which produce ComparisonExpression and ConjunctionExpression objects.
• Iteration: Calling for x in promise: or iter(promise) raises ValueError("... is a Promise. Promise objects are not iterable").
• Item and Attribute Access: Indexing (promise["key"], promise[0]) or attribute access (promise.field) appends to attr_path on the promise's NodeOutput to resolve nested values on schematized structs and dictionaries dynamically.

Task Outputs vs. create_node

Direct task invocation and explicit node creation with create_node serve different purposes in workflow graph assembly.

Direct Task Invocations

When you call a task directly inside a workflow, you receive a Promise (or namedtuple of promises):

@task
def fetch_data() -> (int, str):
    return 100, "dataset"


@workflow
def data_flow() -> (int, str):
    data_out = fetch_data()
    # data_out is a namedtuple of Promise objects (e.g., data_out.o0, data_out.o1)
    # It has NO .outputs attribute.
    return data_out.o0, data_out.o1

Explicit Node Creation with create_node

When you need direct access to the underlying Node instance—such as when sequencing tasks without passing data, using ImperativeWorkflow, or looking up output names dynamically—call create_node:

from flytekit import create_node, task, workflow


@task
def setup():
    print("Setting up storage")


@task
def process() -> str:
    return "done"


@workflow
def manual_node_flow() -> str:
    # create_node returns a Node instance
    setup_node = create_node(setup)
    proc_node = create_node(process)

    # Order execution: setup_node runs before proc_node
    setup_node >> proc_node

    # Access outputs via named attributes (e.g. proc_node.o0)
    # or via the proc_node.outputs dictionary
    return proc_node.outputs["o0"]

Key Differences Between Outputs

Dimension	p = task(...)	node = create_node(task, ...)
Return Type	Promise, VoidPromise, or namedtuple	Node
node.outputs Access	Raises AttributeError (Promise has no .outputs)	Returns dict[str, Promise]
Named Attribute Access	p.o0, p.field (returns Promise)	node.o0, node.my_named_output (returns Promise)
Use Case	Standard functional workflow wiring	Non-data dependencies, imperative workflows, dynamic name lookups

Attempting to access .outputs on a Node instance that was not initialized through create_node raises an AssertionError("Cannot use outputs with all Nodes, node must've been created from create_node()").

Programmatic Composition with ImperativeWorkflow

create_node and .outputs are central when defining workflows programmatically using flytekit.Workflow (ImperativeWorkflow):

from flytekit import Resources, Workflow, task


@task
def step_one(a: int) -> str:
    return str(a)


@task
def step_two(b: str):
    print(f"Consumed {b}")


# Construct workflow imperatively
wb = Workflow(name="imperative_pipeline")
wb.add_workflow_input("input_val", int)

# Add entities and wire outputs via dictionary lookups
node1 = wb.add_entity(step_one, a=wb.inputs["input_val"]).with_overrides(
    requests=Resources(cpu="1")
)
node2 = wb.add_entity(step_two, b=node1.outputs["o0"])

node1 >> node2
wb.add_workflow_output("final_result", node1.outputs["o0"])

Per-Node Overrides and Dependency Sequencing

Workflows can override execution settings on individual nodes without modifying task definitions.

Applying Overrides

Both Node and unresolved Promise objects expose .with_overrides(...). When invoked on a Promise, the method delegates the override call directly to self.ref.node.with_overrides(...).

import datetime
from flytekit import Cache, Resources, task, workflow


@task(cache=False)
def train_model(epochs: int) -> str:
    return f"model-v1-{epochs}"


@workflow
def training_workflow(epochs: int = 10) -> str:
    # Applying overrides to resources, timeouts, retries, and caching
    train_promise = train_model(epochs=epochs).with_overrides(
        node_name="heavy-trainer-node",
        requests=Resources(cpu="4", mem="8Gi"),
        limits=Resources(cpu="8", mem="16Gi"),
        timeout=datetime.timedelta(hours=2),
        retries=3,
        interruptible=True,
        cache=Cache(version="2.0", serialize=True),
    )
    return train_promise

Supported Node Override Parameters

• node_name (str): Replaces the generated node ID with a custom DNS-compliant identifier.
• requests and limits (Resources): Configures compute bounds (cpu, mem, gpu, ephemeral_storage).
• resources (Resources): A single unified resource specification. Note: Setting resources simultaneously with requests or limits raises a ValueError.
• timeout (int | datetime.timedelta): Sets maximum task execution duration.
• retries (int): Configures a RetryStrategy override.
• interruptible (bool): Determines whether the node executes on spot/interruptible instances.
• cache (bool | Cache): Configures caching behavior. When passing a Cache object, cache.version must be provided (must specify cache version when overriding).
• container_image (str): Overrides the Docker image used to execute the node.
• accelerator (BaseAccelerator): Specifies GPU accelerator types.
• shared_memory (str | True): Attaches an /dev/shm shared memory volume.
• pod_template (PodTemplate): Specifies Kubernetes PodTemplate specifications for this node.

All values passed into .with_overrides(...) must be static compile-time constants. Passing a dynamic Promise to any metadata parameter (such as retries, node_name, or requests) raises an AssertionError via assert_not_promise(...).

Non-Data Dependency Sequencing

When tasks must execute in sequence without passing outputs, use the right-shift operator (>>) or .runs_before():

@task
def clean_database():
    print("Cleaned")


@task
def populate_database():
    print("Populated")


@workflow
def sequence_workflow():
    clean_node = create_node(clean_database)
    pop_node = create_node(populate_database)

    # clean_node runs before pop_node
    clean_node >> pop_node
    # Equivalent to: clean_node.runs_before(pop_node)

The right-shift operator can also be chained across promises directly: task_a() >> task_b() >> task_c().

Failure Handlers (on_failure)

The @workflow decorator accepts an on_failure parameter to specify a cleanup task or subworkflow when an upstream node fails.

Interface Requirements for Failure Handlers

Flytekit enforces strict validation between workflow inputs and the on_failure entity inputs during workflow compilation:

1. Input Superset: The failure handler must accept all inputs declared by the parent workflow ((failure_node_inputs | workflow_inputs) == failure_node_inputs).
2. Optional Extra Inputs: Any parameter in the failure handler that is not present in the parent workflow signature must be typed with typing.Optional[...] (e.g., Optional[FlyteError] = None).

Violating either constraint raises FlyteFailureNodeInputMismatchException during workflow construction.

import typing
from flytekit import task, workflow
from flytekit.exceptions.system import FlyteError


@task
def clean_up_resources(
    cluster_id: str,
    region: str,
    err: typing.Optional[FlyteError] = None,
):
    print(f"Cleaning up cluster {cluster_id} in {region}")
    if err:
        print(f"Failed at node {err.failed_node_id}: {err.message}")


@task
def provision_node(cluster_id: str, region: str):
    print(f"Provisioning {cluster_id} in {region}")


@task
def failing_step():
    raise RuntimeError("Compute step crashed")


# Workflow passes cluster_id and region to on_failure automatically
@workflow(on_failure=clean_up_resources)
def cluster_lifecycle_wf(cluster_id: str, region: str = "us-west-2"):
    p = create_node(provision_node, cluster_id=cluster_id, region=region)
    f = create_node(failing_step)
    p >> f

Error Capture During Local and Remote Execution

When a workflow fails:

• Remote Execution (Flyte Cluster): The Flyte engine intercepts the node failure and automatically dispatches the failure node (_failure_node with ID DEFAULT_FAILURE_NODE_ID), binding the original workflow inputs and injecting execution failure context into the task container.
• Local Workflow Execution: When an exception occurs inside a local workflow execution, flytekit's internal handler inspects the failure entity interface. If an err parameter exists in the failure entity's signature, flytekit instantiates a FlyteError(failed_node_id=failed_node.id, message=str(exc)) and passes it to the handler before re-raising the original exception.