Skip to main content

Workflow composition, failure handlers, and nodes

Compose workflows with task calls​

When a workflow body calls a task, the result is not an ordinary Python value while flytekit is compiling the workflow. Use the task result as a workflow value—pass it to another entity, unpack declared outputs, or return it from the workflow—rather than using it in Python operations such as range() or truth-value testing.

@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 workflow decorator creates a PythonFunctionWorkflow. During compilation, PythonFunctionWorkflow.compile executes the function body in a compilation context. The common flyte_entity_call_handler dispatches each task or workflow call to node creation, so t1(a=a) contributes a graph node and returns output promises. create_task_output preserves the entity interface: a single unnamed output is returned as one Promise; multiple outputs are returned in a named-tuple-like wrapper, which is why the example can unpack x, y and _, v.

A compiled output promise is a Promise whose ref is a NodeOutput. NodeOutput records the originating node, output variable, and optional attribute path. A Promise also supports promise-aware attribute and index access (result.field and result[0]), comparisons, dependency chaining, and with_overrides. An unresolved promise cannot be evaluated with eval(), iterated, or tested for truth: Promise.__bool__ raises ValueError. Use flytekit's conditional/comparison mechanisms instead of Python and/or or direct truth testing.

At local execution time, the same dispatcher uses the entity's local execution path. The workflow machinery converts inputs to Flyte literals, executes the graph, validates the declared output count, and repackages results as promises or VoidPromise values as appropriate. Thus, the workflow source uses one call style for compilation and local execution, while the object behind the call differs by context.

WorkflowFailurePolicy.FAIL_IMMEDIATELY is the default workflow policy; the example selects FAIL_AFTER_EXECUTABLE_NODES_COMPLETE. The two policies are serialized by WorkflowMetadata as the supported Flyte workflow metadata values.

Build explicit nodes​

Use create_node when you need a Node object—for example, to express ordering between entities that do not exchange data, or to bind outputs explicitly in an ImperativeWorkflow. Its inputs are keyword-only.

# 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"])

ImperativeWorkflow.add_entity delegates to create_node. In compilation mode, create_node calls the entity to construct the underlying graph node, initializes that node's output dictionary, and attaches each output both by name (node.outputs["o0"]) and as an attribute (node.o0). add_workflow_output then converts the selected promise/output reference into the workflow's output binding.

The equivalent function-style composition returns the task result directly:

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

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

Do not interchange these APIs. x = t1(a=in1) is a Promise or an output wrapper created by create_task_output; it is not a Node and does not expose .outputs. Conversely, create_node(t1, ...) returns a Node in compilation mode, and Node.outputs intentionally raises AssertionError for nodes not created through create_node:

@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

For side-effect-only entities, create nodes and add an explicit edge:

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

t2_node.runs_before(t1_node)
# OR
t2_node >> t1_node

Node.runs_before adds the current node to the other node's upstream list, and Node.__rshift__ returns the other node so chains can be written. You can bind inputs and override the resulting node in the same expression:

t3_node = create_node(t3, in1=some_int)
t3_node = create_node(t3, in1=some_int).with_overrides(...)

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

create_node is intended for workflow or dynamic-task compilation/local-execution contexts and validates the entity type. Calling it with positional inputs raises flytekit's argument validation error. For a void entity, it returns a node without outputs. For a single-output entity, local execution still uses the interface output wrapper; consume the named output (node.o0 or node.outputs["o0"]) rather than assuming a bare Python value.

Override an individual node​

Apply an override either to an explicitly created node or to an unresolved task-call result:

@workflow
def my_wf(x: typing.List[int]) -> typing.List[typing.Optional[str]]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

Promise.with_overrides forwards the call to self.ref.node.with_overrides while the promise is unresolved, then returns the promise/output wrapper. Node.with_overrides mutates the referenced node and returns that same node. Supported settings include node_name, aliases, requests, limits, resources, timeout, retries, interruptible, task_config, container_image, accelerator, cache, shared_memory, and pod_template.

Examples of the scalar/resource forms are:

node.with_overrides(
    node_name="preprocess",
    timeout=60,
    retries=2,
    interruptible=True,
    container_image="image:tag",
)

Timeout integers are interpreted as seconds; datetime.timedelta is also accepted, and timeout=None clears the timeout. node_name is DNS-normalized. aliases must be a dictionary. requests and limits must be Resources; use resources instead of combining it with either requests or limits. Supplying requests without limits emits a warning and clamps requests to the original limits. Resource values and override parameters such as retries, image, and cache settings cannot be promises. A Cache override must include a cache version, and deprecated cache parameters cannot be supplied alongside a Cache object. Task-configuration overrides preserve the original task-config type.

Run cleanup after workflow failure​

Attach a failure handler when a failed workflow must invoke cleanup, while still reporting the workflow as failed. The handler signature is strict: it must accept every workflow input, and every additional handler input must be Optional.

@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 t1(a: int, b: str):
    print(f"{a} {b}")
    raise ValueError(error_message)

@task
def delete_cluster(name: str, err: typing.Optional[FlyteError] = None):
    print(f"Deleting cluster {name}")
    print(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

with pytest.raises(ValueError):
    wf()

During PythonFunctionWorkflow.compile, flytekit validates the handler against the workflow interface: each workflow input must occur in the handler signature, while an extra required input raises FlyteFailureNodeInputMismatchException. The special err input is populated with a FlyteError describing the failed node and exception message. The handler is compiled separately as an out-of-band failure node and must produce exactly one task or workflow node; it is removed from the main workflow node list.

At runtime, WorkflowBase.__call__ invokes the handler when execution raises, then re-raises the original exception. Cleanup therefore does not turn the failed workflow into a successful one, and an exception raised by the handler is not specially swallowed. ImperativeWorkflow exposes the corresponding programmatic registration method, add_on_failure_handler; python_function_task.get_as_workflow uses it after mapping the imperative workflow's inputs and outputs.

Output and dependency edge cases​

A task or workflow with no declared outputs returns a VoidPromise. It supports dependency ordering and overrides, but value operations and comparisons raise assertions. A workflow with no declared outputs must return None/VoidPromise; returning a value is rejected, while a declared output with no returned value is also rejected.

Finally, use .outputs only for explicit create_node results. Ordinary task calls expose a single Promise or a named output wrapper, and promises carry NodeOutput references until execution. That distinction is what lets function-style workflows compose naturally while imperative workflows retain explicit node/output maps.