Workflow composition, failure handlers, and nodes

The promise model: why a workflow body doesn't return Python values

A @workflow-decorated function looks like ordinary Python, but its body is not executed the way you'd expect. When flytekit scans the body, a = t1(...) does not give you the task's output value — it gives you a Promise. Trying to range(a) on it raises a ValueError, because Promise objects are deliberately not iterable:

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


@workflow
def wf(a: int) -> int:
    return t1(a=a)

The reason is documented directly in the workflow decorator: "when flytekit scans the workflow function, the objects being passed around between the tasks are not your typical Python values." A workflow's body runs at serialization time to express the DAG structure (which task feeds which), not to compute values. Each task invocation materializes into a Node, and the values you thread between calls are Promise objects pointing at that node's not-yet-known outputs.

Promise (flytekit/core/promise.py) wraps one of two things, and its docstring lays out the dual nature:

• A ready Promise holds an actual Flyte Literal — is_ready is True, and .val is the resolved value. This happens during local execution, where the same function body is re-run for testing and produces real values.
• An unready Promise holds a NodeOutput reference to an upstream node's output — is_ready is False, and .ref points at that NodeOutput. This is what compilation produces.

@property
def is_ready(self) -> bool:
    """Returns if the Promise is READY (is not a reference and the val is actually ready)"""
    return self._promise_ready

When a task is invoked from inside a workflow, three things may happen, as flyte_entity_call_handler (promise.py:1442) decides by inspecting the current context:

1. Compilation — ctx.compilation_state is set, so it calls create_and_link_node(ctx, entity=entity, **kwargs) to build a node and return Promises pointing at it.
2. Local workflow execution — the context is in a local-execution state, so it runs local_execute and wraps results so later tasks know how to unwrap them.
3. Plain local call — no compilation, no execution state, so it runs the task and returns native Python values.

Every t1(...) call in a workflow funnels through this handler via WorkflowBase.__call__ (workflow.py:315) and PythonTask.__call__ / LaunchPlan.__call__.

The same body is thus executed twice: once during compile() to build the DAG, and once during local execution to exercise it. create_task_output (prompt.py:772) is what packages the returned Promises — a single Promise for one output, or a collections.namedtuple of Promises for multiple. That's also why a one-element NamedTuple produces a one-element tuple rather than a bare Promise, and why on the local-execution path create_native_named_tuple reconstructs native values.

Composing workflows with workflow()

workflow() (flytekit/core/workflow.py) is a decorator that turns a function into a PythonFunction petition. It accepts two overloads — bare @workflow and parameterized @workflow(...) — and several parameters:

def workflow(
    _workflow_function=None,
    failure_policy=None,
    interruptible=False,
    on_failure=None,
    docs=None,
    pickle_untyped=False,
    default_options=None,
):
    """Declare this function to be a Flyte workflow and return the workflow object."""

The two metadata concerns are split into separate dataclasses in the same module:

• WorkflowMetadata holds the on_failure policy — how the workflow behaves when a node fails. WorkflowFailurePolicy (an enum in flytekit/core/workflow.py) gives FAIL_IMMEDIATELY (the default: the whole workflow enters a failed state as soon as a component node fails) and FAIL_AFTER_EXECUTABLE_NODES_COMPLETE (remaining runnable nodes still execute).
• WorkflowMetadataDefaults holds interruptible, described in its docstring as the defaults "handed down to a workflow's tasks" (as opposed to WorkflowMetadata, which is about the workflow itself).

A workflow's body flows through PythonFunctionWorkflow.compile() (workflow.py:822), which:

1. constructs input Promises (each a NodeOutput pointing at the shared GLOBAL_START_NODE) via construct_input_promises,
2. calls the decorated function with those Promises as kwargs — this triggers all the create_and_link_node calls and builds the node list,
3. validates the on-failure handler,
4. converts the function's return values into _output_bindings via binding_from_python_std.

input_kwargs = construct_input_promises([k for k in self.interface.inputs.keys()])
input_kwargs.update(kwargs)
workflow_outputs = self._workflow_function(**input_kwargs)
# ...
self._nodes = all_nodes
self._output_bindings = bindings

The class hierarchy is WorkflowBase (abstract holder of name, metadata, interface, nodes, output bindings, and the on-failure entity) → PythonFunctionWorkflow, with ImperativeWorkflow and ReferenceWorkflow as siblings. ImperativeWorkflow lets you assemble the same DAG programmatically without a decorated body — you call add_workflow_input, add_entity (which delegates to create_node), add_workflow_output, and add_on_failure_handler (all in flytekit/core/workflow.py):

wb = ImperativeWorkflow(name="my_workflow")
wb.add_workflow_input("in1", str)
node = wb.add_entity(t1, a=wb.inputs["in1"])
wb.add_entity(t2)
wb.add_workflow_output("from_a", node.outputs["o0"])

Accessing a manually-created node's outputs

create_node (flytekit/core/node_creation.py) is the escape hatch for when you need to control node ordering or access a node object directly, rather than relying on the implied ordering of an ordinary task call. Its docstring leads with the use case: "if you have t1() and t2(), both of which do not take in nor produce any outputs, how do you specify that t2 should run before t1?" — via t2_node.runs_before(t1_node) or the >> operator.

create_node accepts only keyword arguments. Positional args raise FlyteAssertion ("Only keyword args are supported to pass inputs to workflows and tasks"). It works in two modes:

def create_node(entity, *args, **kwargs):
    """Create a node for a callable Flyte entity."""
    if len(args) > 0:
        raise _user_exceptions.FlyteAssertion(
            f"Only keyword args are supported to pass inputs to workflows and tasks."
        )

• Compilation mode (inside a workflow or dynamic task): it calls the entity, which routes, and drops a new node onto ctx.compilation_state.nodes[-1] — the most recently added node. create_node then sets node._outputs = {} and attaches each output both as a node.outputs[name] dict entry and as an attribute:

setattr(node, output_name, attr)
node.outputs[output_name] = attr

For a single output, the Promise is attached under its name directly; for a custom named tuple, each named field becomes an attribute. The returned object is the Node itself (or a VoidPromise when the entity produces nothing). Inside a @workflow, you then consume those output Promises:

@workflow
def my_wf(a: str) -> typing.Tuple[str, typing.List[str]]:
    t1_node = create_node(t1, a=a)
    dyn_node = create_node(my_subwf, a=3)
    return t1_node.o0, dyn_node.o0

• Local execution (running the workflow locally): it calls the entity and returns native results, but still "tupletizes" a single output through entity.python_interface.output_tuple(results) so the local path matches the named-tuple convention.

Node.outputs is only populated for nodes created through create_node. A Node built implicitly by a task call has _outputs = None, so accessing .outputs raises:

@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

This is the core distinction explored in the last section below.

Per-node overrides with with_overrides

Both Node and Promise expose with_overrides(...) for per-node execution settings. On a Promise from an ordinary task call, calling .with_overrides(...) forwards to the referenced node's method:

# On Promise
if not self.is_ready:
    self.ref.node.with_overrides(
        node_name=node_name, aliases=aliases, requests=requests, limits=limits, timeout=timeout,
        retries=retries, interruptible=interruptible, name=name, task_config=task_config,
        container_image=container_image, accelerator=accelerator, cache=cache,
    )
return self

Node.with_overrides (flytekit/core/node.py) mutates the node in place and returns self, so you can chain it. The full parameter set:

def with_overrides(
    self,
    node_name=None,
    aliases=None,
    requests=None,
    limits=None,
    timeout=TIMEOUT_OVERRIDE_SENTINEL,
    retries=None,
    interruptible=None,
    name=None,
    task_config=None,
    container_image=None,
    accelerator=None,
    cache=None,
    shared_memory=None,
    pod_template=None,
    resources=None,
):
    """Apply execution overrides to this node."""
    return self

A few behaviors worth knowing, all enforced inside the method:

• timeout is guarded by Node.TIMEOUT_OVERRIDE_SENTINEL. If you pass None, the node's timeout becomes datetime.timedelta(); an int becomes datetime.timedelta(seconds=timeout); a timedelta is used directly; and any other type raises ValueError.
• resources cannot be combined with requests/limits. Passing both raises ValueError ("resource" should not be used together with "limits" or "requests"). When only resources is given, it is split into requests and limits. If you set requests without limits, flytekit logs a warning that requests are clamped to original limits.
• cache accepts either a bool or a Cache object. Passing a True bool becomes Cache(serialize=..., ignored_inputs=...). Passing a Cache requires version (else ValueError), and the deprecated cache_serialize/cache_version/cache_ignore_input_vars kwargs still parse for backward compatibility but raise if used alongside a Cache.
• node_name is DNS-ified (_dnsify(node_name)).
• aliases — with_overrides(aliases={"output_name": "alias"}) adds Alias entries to the node; it must be a dict[str, str], anything else raises AssertionError.

Many override values cannot themselves be Promises — assert_not_promise guards retries, interruptible, cache, node_name, container_image, shared_memory, and pod_template, and the same guard runs when building extended resources for accelerator/shared_memory.

run_entity on a Node unwraps map-task wrappers: for a MapPythonTask it returns the wrapped run_task, for an ArrayNodeMapTask the underlying python_function_task, and otherwise the flyte_entity itself.

Ordering dependencies without data flow

Data-flow between nodes creates implicit upstream dependencies. When two tasks exchange no data but must still run in a specific order, flytekit gives you two equivalent spellings:

t2_node = create_node(t2)
t3_node = create_node(t3)
t2_node >> t3_node        # node_1 >> node_2 == node_1.runs_before(node_2)
t3_node.runs_before(t2_node)

Node.__rshift__ calls runs_before, which appends self to the other node's _upstream_nodes:

def runs_before(self, other: Node):
    """Add this node to the other node's upstream list."""
    if self not in other._upstream_nodes:
        other._upstream_nodes.append(self)

def __rshift__(self, other: Node):
    self.runs_before(other)
    return other

>> is chosen over << because it's what most users are familiar with. Tasks with no outputs return a VoidPromise, and VoidPromise.__rshift__ handles ordering the same way while rejecting any value-like operation — comparisons, arithmetic, even str() raise AssertionError naming the task.

On-failure handlers

The on_failure parameter of @workflow attaches a failure handler. Pass any Task or another WorkflowBase:

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

@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

The handler signature must contain every workflow input

When compile() validates the handler (_validate_add_on_failure_handler, workflow.py:789), it enforces a strict contract: the handler's inputs must be a superset of the workflow's inputs, and any additional inputs must be Optional:

workflow_inputs = self.python_interface.inputs
failure_node_inputs = self.on_failure.python_interface.inputs

if (failure_node_inputs | workflow_inputs) != failure_node_inputs:
    raise FlyteFailureNodeInputMismatchException(self.on_failure, self)
additional_keys = failure_node_inputs.keys() - workflow_inputs.keys()
for k in additional_keys:
    if not is_optional_type(failure_node_inputs[k]):
        raise FlyteFailureNodeInputMismatchException(self.on_failure, self)

So clean_up(name, err=None) is valid because name matches the workflow input and err is Optional. A handler with a required extra argument — or one that omits a workflow input — raises FlyteFailureNodeInputMismatchException. ImperativeWorkflow.add_on_failure_handler (workflow.py:685) runs the same validation.

The special err parameter gets the failure detail injected. During local execution, Workflow.__call__ catches an exception from the body and, if the handler's interface contains "err", builds a FlyteError(failed_node_id=..., message=str(exc)) and calls the handler with all the original inputs plus err:

if self.on_failure:
    if self.on_failure.python_interface and "err" in self.on_failure.python_interface.inputs:
        id = self.failure_node.id if self.failure_node else ""
        input_kwargs["err"] = FlyteError(failed_node_id=id, message=str(exc))
    self.on_failure(**input_kwargs)
raise exc

If the handler doesn't declare err, no error is injected. The compiled failure node is stored as self._failure_node and given the reserved id DEFAULT_FAILURE_NODE_ID ("efn"). In the imperative path, add_on_failure_handler calls create_node and then pops that node off the compilation state (ctx.compilation_state.nodes.pop(-1)) — in the docstring's words, failure nodes are special because "we don't want them to be part of the main workflow."

Promises vs. create_node(...).outputs

Now that both access paths are visible, the distinction is precise:

• An ordinary task call (x = t1(...)) returns a Promise (or a tuple of Promises) directly. You pass x downstream or access attributes on it (x.attr, x[0]) — those append to the promise's attribute path. But .outputs on a Promise is meaningless; Promise has no .outputs property. The Promise is the handle to the data flowing between nodes.
• create_node(t1, ...) returns the Node itself, and the Node is the accessor for the manual pattern. Its .outputs dict and its output attributes (node.o0, node.<named_output>) let you pull output Promises off the node after creation, which is what the imperative workflow pattern relies on (add_workflow_output("from", node.outputs["o0"])).

Conceptually: a Promise is a value a node will produce, while a Node's .outputs is a lookup index of the values a node will produce. Both resolve to the same underlying Promise, but they're reached differently. Because create_node attaches its outputs as attributes, the functional pattern reads naturally:

t1_node = create_node(t1, a=a).with_overrides(aliases={"t1_output": "foo"})
t2_node = create_node(t2, a=[t1_node.t1_output, b])
return t1_node.t1_output, t2_node.o0

while the imperative pattern favors the dict accessor since the output name is usually held in a string variable.

The following table summarizes what each object exposes:

Object	Location	Key attributes / meaning
Promise	flytekit/core/promise.py	A resolvable value from a node: is_ready, .val, .ref, .var, .attr_path, eval(), with_overrides(), >>, comparison ops (→ ComparisonExpression), [i]/.attr indexing
Node	flytekit/core/node.py	An element of a compiled workflow DAG: runs_before(), with_overrides(), .bindings, .upstream_nodes, .flyte_entity, and .outputs/.o0 only via create_node
VoidPromise	flytekit/core/promise.py	Returned by a no-output entity; supports ordering (runs_before, >>) only, all other ops raise
NodeOutput	flytekit/core/promise.py	The value a particular node+var produces; .node, .node_id, .var, .with_attr()

A final caveat while composing: Promise equality (==, !=) builds a ComparisonExpression for use inside a conditional, it does not compare a value. And you can't bool() a Promise (via if promise:) or chain it with and/or — those raise ValueError, with the docstring pointing you to &/| or the is_true()/is_false()/is_none() helpers.