Workflow composition, failure handlers, and nodes

Function-style workflow composition

When a workflow calls a task during compilation, the assigned results are not ordinary Python values. They are Promise objects that describe bindings to the task node. Write the workflow as dataflow: pass task outputs into later calls and return the outputs that should become workflow outputs.

The workflow decorator documentation includes this complete composition pattern:

@task
def t1(a: int) -> typing.NamedTuple("OutputsBC", [("t1_int_output", int), ("c", str)]):
    a = a + 2
    return a, "world-" + str(a)

@workflow(interruptible=True, failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def wf(a: int) -> typing.Tuple[str, str]:
    x, y = t1(a=a)
    _, v = t1(a=x)
    return y, v

The first call produces two promises, bound to x and y; the second call consumes x and produces v. PythonFunctionWorkflow.compile() creates promises for the workflow inputs, invokes the workflow function under a compilation context, collects the nodes created by those calls, and converts the returned promises into output bindings. The workflow metadata records the selected WorkflowFailurePolicy; the supported values are FAIL_IMMEDIATELY and FAIL_AFTER_EXECUTABLE_NODES_COMPLETE.

At runtime, WorkflowBase.local_execute() converts native inputs to Flyte literals and normalizes the result. The call machinery in flyte_entity_call_handler() explicitly has two modes: compilation creates a node and returns promises, while local execution runs the entity and wraps its outputs so operations such as overrides can still be attached. Consequently, Python expressions that require a concrete value—such as using a task output as the argument to range()—do not generally work in a compiled workflow.

A task with no declared outputs returns a VoidPromise. It can be used for ordering and overrides, but it cannot be passed as a downstream input; promise binding raises an assertion identifying the task that produced no outputs. Value operations and comparisons on VoidPromise also raise assertions.

Ordering nodes explicitly

Data dependencies normally establish ordering automatically: when a Promise is bound as an input, binding_data_from_python_std() records the promise's source node as an upstream node. For tasks without data outputs, or when you need an explicit edge, use the node ordering API:

create_cluster_node = create_node(create_cluster, name="flyteorg")
t1_node = create_node(t1, a=1, b="2")
delete_cluster_node = create_node(delete_cluster, name="flyteorg")

create_cluster_node >> t1_node >> delete_cluster_node

Node.__rshift__() calls runs_before() and returns the downstream node, so the expression can be chained. runs_before() appends the upstream node to the other node's upstream list only when it is not already present.

Explicit node construction with create_node

Use create_node() when you need the compiled Node itself—for example, to create a dependency-only node, apply a node-level override, or refer to outputs by their declared names. Inputs to create_node() are keyword-only; positional inputs raise FlyteAssertion.

t3_node = create_node(t3, in1=some_int).with_overrides(timeout=60, retries=2)

In compilation mode, create_node() calls the entity through the promise/linking machinery, retrieves the node added to the compilation state, initializes its output mapping, and exposes each output in two forms:

t4_node = create_node(t4)
t5(in1=t4_node.o0)

output_name = "o0"
t5(in1=t4_node.outputs[output_name])

The attribute form (t4_node.o0) and mapping form (t4_node.outputs["o0"]) refer to the node's output promise. The mapping is useful when the output name is held in a variable. A void entity returns a node without output entries.

create_node() is context-sensitive. It is intended for a valid compilation or local-execution context, is rejected while a skipped conditional branch is being evaluated, and cannot run a remote entity during local execution. During local execution it calls the entity normally rather than returning a compiled Node; its output handling therefore differs from compilation, including tuple-wrapping a single result and preserving named-tuple behavior.

create_node(...).outputs is not a task-call promise

Do not write t4(...).outputs for an ordinary task call. A normal task or workflow call is handled by flyte_entity_call_handler() and returns a Promise, a tuple/named tuple of promises, or a VoidPromise. It does not return a Node with an outputs mapping.

The Node.outputs property makes this distinction explicit:

@property
def outputs(self):
    if self._outputs is None:
        raise AssertionError("Cannot use outputs with all Nodes, node must've been created from create_node()")
    return self._outputs

Only nodes initialized by create_node() expose that mapping. ImperativeWorkflow.add_entity() uses the same explicit-node path, which is why imperative workflow construction can bind outputs by name:

node = wb.add_entity(self, **input_kwargs)
for output_name, output_python_type in self.python_interface.outputs.items():
    wb.add_workflow_output(output_name, node.outputs[output_name])

For function-style composition, instead assign the ordinary task-call result and return it from the workflow. For explicit composition, use the Node returned by create_node() or add_entity() when you need named output references.

Per-node overrides

Apply execution settings to the node that will run by calling with_overrides() on a Node, or call it on an incomplete Promise to delegate to the producing node:

node = create_node(t3, in1=some_int).with_overrides(
    node_name="preprocess",
    timeout=60,
    retries=2,
    interruptible=True,
    container_image="python:3.11",
)

Node.with_overrides() mutates the node. It supports DNS-normalized node_name, aliases, requests and limits (or the combined resources argument), timeout as integer seconds or datetime.timedelta, retries, interruptibility, cache settings, task configuration, container image, accelerator, shared memory, and pod template. Promise.with_overrides() forwards these settings to self.ref.node when the promise is incomplete and then returns the promise.

Keep the override values concrete. The implementation rejects promise-valued metadata such as retries, interruptibility, cache version, and container image. Do not combine resources with requests or limits; with_overrides() raises ValueError and asks you to set only resources. If you provide requests without limits, flytekit logs a warning that requests are clamped to the original limits. A Cache used for an override must include an explicit cache version, and deprecated cache parameters cannot be combined with a Cache object.

Failure handlers

A failure handler is a separate task or workflow invoked after a workflow call raises. Its interface must include every workflow input. It may add inputs only when those additional inputs are annotated as Optional; a required extra input fails compilation with FlyteFailureNodeInputMismatchException.

This handler is valid because it accepts the workflow input name and declares the additional err input as optional:

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name} due to {err}")

@workflow(on_failure=clean_up)
def wf(name: str = "flyteorg"):
    create_cluster(name=name)

The handler name err matters for error injection. At runtime, WorkflowBase.__call__() catches the workflow exception, and when the failure handler interface contains err, it supplies a FlyteError containing the failed node ID and the exception message. It then calls the handler and re-raises the original exception; the handler does not replace the workflow's original failure.

For explicit ordering, the source uses the following pattern:

@task
def create_cluster(name: str):
    print(f"Creating cluster: {name}")

@task
def delete_cluster(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name}")
    print(err)

@task
def t1(a: int, b: str):
    print(f"{a} {b}")
    raise ValueError(error_message)

@workflow(on_failure=clean_up)
def wf(name: str = "flyteorg"):
    c = create_cluster(name=name)
    t = t1(a=1, b="2")
    d = delete_cluster(name=name)
    c >> t >> d

During compilation, WorkflowBase.add_on_failure_handler() validates the handler against the workflow input promises, compiles it in a separate context, and removes its node from the main node list. The result is stored as failure_node; compilation requires the handler to produce exactly one task or workflow node. At runtime, the normal graph and failure path therefore have distinct roles.

The composition lifecycle is:

entity call
  -> Promise input binding and upstream-node discovery
  -> Node creation and graph links
  -> workflow output bindings

on_failure handler
  -> separate validation/compilation
  -> failure_node (not a normal workflow node)
  -> invocation with workflow inputs and optional FlyteError