Workflow composition, failure handlers, and nodes

Compose workflows as graphs

A function decorated with @workflow is evaluated while flytekit compiles the workflow, not as an ordinary runtime function whose task results are immediately available to Python. PythonFunctionWorkflow.compile() creates input promises, invokes the workflow function under a CompilationState, collects the nodes generated by task and workflow calls, validates the failure handler, and then builds workflow output bindings (workflow.py, PythonFunctionWorkflow.compile). Write the workflow body in terms of entity calls and their outputs:

import typing

from flytekit import task, workflow
from flytekit.core.workflow import WorkflowFailurePolicy


@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

This is the composition pattern used in the test_workflow_values example embedded in workflow.py. t1(a=x) receives the first task's output as a promise, so flytekit can create the corresponding input binding and upstream-node relationship. The failure_policy in the example is not the default: WorkflowFailurePolicy.FAIL_IMMEDIATELY is the default, while FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows other runnable nodes to finish before the workflow fails. The same example sets the workflow's default interruptibility with interruptible=True.

The function body is nevertheless executable locally. WorkflowBase.__call__ compiles outside eager execution, invokes flyte_entity_call_handler, and invokes on_failure if that call raises. WorkflowBase.local_execute() supplies literal-backed values and repackages the local results according to the workflow interface. The dual behavior is why a value such as x must be treated as a flytekit workflow value while the graph is being composed, rather than as an ordinary integer on which arbitrary Python control flow can operate.

Imperative composition

Use ImperativeWorkflow (also exposed as Workflow in the documented example) when you want to name inputs and outputs explicitly instead of returning them from a decorated function:

# Create the workflow with a name.
wb = Workflow(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_n0t1", node.outputs["o0"])

This is the imperative example in workflow.py. add_entity() enters compilation mode and delegates to explicit node creation; add_workflow_output() plays the role of the function's return statement. The function-style equivalent uses a named tuple when the workflow output needs a particular name:

nt = typing.NamedTuple("wf_output", [("from_n0t1", str)])


@workflow
def my_workflow(in1: str) -> nt:
    x = t1(a=in1)
    t2()
    return nt(x)

The side-effect-only t2() is composed independently in both examples. If an ordering edge is required, add it explicitly as described below.

Nodes and explicit ordering

There are two different APIs that can appear to “call a task.” A normal task or workflow call goes through create_and_link_node() in promise.py. That function constructs a Node, registers it in the active compilation state, and returns output objects: a VoidPromise for an entity with no outputs, a Promise for one output, or the entity's named tuple of promises for multiple outputs. The returned object is therefore not the graph Node.

create_node() in node_creation.py is the explicit/manual API. It accepts a task, workflow, launch plan, or remote entity and only accepts keyword inputs:

from flytekit.core.node_creation import create_node


t3_node = create_node(t3, in1=some_int)
t4_node = create_node(t4)
t5(in1=t4_node.o0)

During compilation, create_node() invokes the entity, obtains the newly appended node, and attaches each output promise both as an attribute and in the node's output dictionary. Thus t4_node.o0 and t4_node.outputs["o0"] refer to the output promise in this explicit-node form. The imperative example uses the dictionary form because output names can be held in a string.

For ordinary calls, use the returned promise or named output field instead:

x, y = t1(a=a)
_, v = t1(a=x)

Do not write t1(a=a).outputs: ordinary task-call results are promises (or a named tuple of promises), and Node.outputs is intentionally unavailable on an ordinary node. Its property raises AssertionError unless _outputs was installed by create_node() (node.py, Node.outputs). Conversely, create_node() returns local execution results rather than a graph node when the context is local. Its source specifically preserves named tuples and wraps a single local result using the entity's output tuple, so the local form of a single-output explicit call still uses the output wrapper described in node_creation.py.

Explicit nodes are particularly useful when tasks do not consume or produce data but must be ordered. Node.runs_before() appends the current node to the other node's upstream-node list if it is not already present. The right-shift operator is equivalent and returns its right-hand operand:

from flytekit.core.node_creation import create_node


t1_node = create_node(t1)
t2_node = create_node(t2)

t2_node.runs_before(t1_node)
# Equivalent shorthand:
t2_node >> t1_node

Data dependencies normally establish upstream nodes automatically: create_and_link_node() discovers nodes in promise bindings and stores them on the newly constructed Node. Use runs_before() or >> when ordering is not expressed by a downstream input, especially for side-effect-only entities. A VoidPromise cannot provide a downstream value; it supports ordering (>>) and node overrides, but value operations raise an assertion.

Manual creation has context-specific restrictions. Positional arguments cause FlyteAssertion; local creation rejects RemoteEntity values and raises when the branch is in BranchEvalMode.BRANCH_SKIPPED. In a local context it executes the entity rather than constructing a graph node.

Promises are the values between nodes

Promise is flytekit's bridge between compilation and local execution. An unresolved promise contains a NodeOutput reference (promise.ref) identifying the producing node and output variable (ref.node, ref.var); a ready promise contains a literal-backed value (promise.val). The object exposes is_ready, var, and attr_path as well. NodeOutput.with_attr() and the promise's __getitem__/__getattr__ methods preserve nested output access by extending that attribute path.

Multiple task outputs retain the names from the entity interface. In the t1 example, x, y = t1(a=a) binds the named tuple's t1_int_output and c promises positionally, while downstream code can use the named result shape when it is retained. create_task_output() creates one Promise per interface output and returns the appropriate output shape; a no-output entity instead produces a VoidPromise.

Do not use a promise as a Python truth value or iterable. Promise.__bool__ raises ValueError, and __iter__ raises because a promise cannot be ranged over. For conditions, use flytekit comparison methods such as is_true(), is_false(), and is_none(), or comparisons combined with & and |; ComparisonExpression and ConjunctionExpression also reject Python truth-value testing.

Per-node overrides

Apply an override to a normal task-call result when you want to change the node created by that call:

x = t1(a=a).with_overrides(node_name="first-t1", timeout=60, retries=2)

An unresolved Promise.with_overrides() forwards the arguments to self.ref.node.with_overrides() and returns the promise. The operation therefore changes the referenced graph node while preserving the value you pass downstream. The same operation can be applied directly to an explicit node:

t3_node = create_node(t3, in1=some_int).with_overrides(node_name="explicit-t3", timeout=60)

Node.with_overrides() mutates the node's settings. Its supported options include node_name, aliases, requests, limits, resources, timeout, retries, interruptible, name, task_config, container_image, accelerator, cache, shared_memory, and pod_template. Names supplied as node_name are DNS-normalized. A resources override cannot be combined with requests or limits; requests without limits produce a warning and are clamped to the original limits. Override metadata and resource values cannot be promises, and task_config must have the same type as the underlying entity's task configuration.

Cache overrides have additional checks. A Cache object must specify a version, and supplying deprecated cache-version or cache-serialization arguments together with a Cache object raises ValueError. The cache=True compatibility path constructs a Cache when no cache version was supplied through the deprecated arguments, while a direct Cache override must have a version. These validations are implemented by Node._override_node_metadata() in node.py.

Failure handlers

A failure handler is itself compiled as a single failure node. Give it every workflow input, then add only optional handler-specific inputs. This is a valid local-execution example from workflow.py:

import typing

from flytekit import task, workflow
from flytekit.types.error.error import FlyteError


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


@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("failure")


@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

Here clean_up accepts the workflow input name and one additional input, err, whose type is Optional[FlyteError]. At local runtime, WorkflowBase.__call__ catches the exception, creates a FlyteError containing the exception text and the failure-node ID, and adds it to the handler call only when the handler interface declares an input named exactly err; it then re-raises the original exception. The embedded example asserts that cleanup prints the generated error and that calling wf() raises ValueError.

Compilation applies the same input contract before creating the failure node. _validate_add_on_failure_handler() compares the workflow and handler interfaces: the workflow input names must be a subset of the handler's input names, and every handler-only input must be optional. Therefore a handler that omits name, or one that adds a required cluster_id: str, raises FlyteFailureNodeInputMismatchException. An optional err is accepted; it receives its runtime FlyteError only under the exact name described above.

The handler is compiled in a nested CompilationState with the workflow's prefix plus "f" (for example, the local assertion in workflow.py observes a failure node ID of fn0). The resulting node is stored as the workflow's failure node and is not treated as an ordinary node in the main workflow body. Compilation requires exactly one generated task or workflow node: no node, or more than one, raises AssertionError with “only either a task or a workflow can be used.” This means a handler should be a single task call or a workflow call that compiles to one node, rather than a body containing multiple independently generated nodes.

The same handler machinery is used by imperative workflows through add_on_failure_handler(). For example, PythonFunctionTask.get_as_workflow() creates an EagerFailureHandlerTask, adds the task as an imperative node, maps each node.outputs[output_name] to a workflow output, and registers the cleanup handler. That integration relies on the explicit-node output dictionary; it does not change the promise-versus-node distinction for ordinary task calls.