Skip to main content

Workflow composition, failure handlers, and nodes

Introduction to Workflow Composition​

A Flyte workflow is a directed acyclic graph (DAG) of tasks where edges are defined by data flow — the output of one task becomes the input of another. When you decorate a function with @workflow, flytekit reads the function body at serialization time (not at remote execution time) to discover how tasks connect.

This distinction matters: the workflow function body is evaluated once when flytekit compiles the DAG structure, not every time the workflow runs on the Flyte platform. This means any plain Python code inside the body — list construction, loops, conditionals — runs during local testing and compilation, but will not execute on the remote Flyte cluster. Only calls to Flyte entities (tasks, sub-workflows, launch plans) become part of the compiled graph.

The @workflow Decorator​

Declare a workflow by decorating a function with @workflow:

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

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

@task
def t2(b: str) -> str:
    return f"hello {b}"

@workflow
def my_wf(a: int, b: str = "world") -> (int, str):
    x = t1(a=a)
    y = t2(b=b)
    return x, y

The decorator accepts these optional parameters:

ParameterTypeDefaultPurpose
failure_policyWorkflowFailurePolicyFAIL_IMMEDIATELYWhether to fail immediately or let remaining runnable nodes complete
interruptibleboolFalseDefault interruptibility for all tasks in this workflow
on_failureWorkflowBase or TaskNoneA handler invoked when the workflow fails
docsDocumentationNoneExtracted from the function docstring if not provided
pickle_untypedboolFalseBypass type checking (not recommended)
default_optionsOptionsNoneDefault labels/annotations for the generated launch plan

In this example, my_wf calls t1 and t2 and returns both results. The call t1(a=a) does not return an integer — it returns a Promise object (explained below). The DAG structure emerges from which outputs feed into which inputs.

Task Outputs and Promises​

When you write x = t1(a=a) inside a workflow, x is not an int — it is a Promise object that represents the future output of the t1 node. Flytekit uses Promise to bridge two modes:

Compilation Mode vs. Local Execution Mode​

During compilation (when flytekit serializes the workflow), Promise.is_ready is False and the Promise.ref property points to a NodeOutput that references the producing node. During local execution, Promise.is_ready is True and Promise.val holds the actual Literal value.

# Inside a workflow body
p = t1(a=5)       # p is a Promise, not an int
print(p.is_ready) # False during compilation, True during local execution

Accessing Nested Outputs​

Promise supports attribute chaining and indexing to access nested outputs:

@task
def get_user() -> UserData:
    return UserData(name="alice", scores=[95, 87])

@workflow
def analyze():
    u = get_user()
    # Attribute access
    name = u.name            # new Promise with attr_path ["name"]
    # Index access
    first_score = u.scores[0]  # new Promise with attr_path ["scores", 0]

Each access calls Promise._append_attr(key), which creates a new Promise with an extended attr_path (and a new NodeOutput with an extended attribute path for remote execution). The original promise is not modified, so you can use it in multiple places.

The VoidPromise​

Tasks that declare no outputs return a VoidPromise instead:

@task
def log_message(msg: str):
    print(msg)

@workflow
def wf():
    result = log_message(msg="hello")  # result is a VoidPromise

VoidPromise rejects nearly all operations — comparison, arithmetic, boolean testing — with an AssertionError message like "Task log_message returns nothing, NoneType return cannot be used". It only supports with_overrides() and __rshift__() for ordering.

Promise Comparisons for Conditionals​

Promises support comparison operators (==, !=, >, <, >=, <=) that return ComparisonExpression objects instead of booleans. Combine expressions with & (AND) and | (OR) to form ConjunctionExpression objects:

@workflow
def wf(x: int):
    # Comparison produces a ComparisonExpression, not a boolean
    expr = (promise_output > 5) & (promise_output < 10)

# But THIS raises ValueError:
# if promise_output:  # ValueError!

Python's and/or keywords trigger __bool__, and both Promise.__bool__ and ComparisonExpression.__bool__ raise ValueError with instructions to use &/| instead. Use conditional() from flytekit.core.condition for workflow branching.

with_overrides on Promises​

Calling with_overrides() on a Promise delegates to the underlying Node.with_overrides(). This lets you chain overrides directly on task calls:

x = t1(a=5).with_overrides(retries=3, timeout=timedelta(minutes=30))

The Node Class and create_node()​

The Node class (flytekit.core.node.Node) is the internal representation of a DAG vertex. Each node stores an id, metadata (NodeMetadata), bindings (input bindings), upstream_nodes (dependency list), and flyte_entity (the task or workflow).

When you call a task normally (t1(a=5)), flytekit creates a node automatically. But some patterns require explicit node creation via create_node():

Explicit Dependency Ordering​

Tasks that don't consume each other's outputs have no data-dependency edge, so flytekit would be free to run them in any order. Use create_node() plus runs_before() (or the >> operator) to enforce sequencing:

from flytekit.core.node_creation import create_node

@task
def t2():
    ...

@task
def t3():
    ...

@workflow
def empty_wf():
    t2_node = create_node(t2)
    t3_node = create_node(t3)
    t3_node.runs_before(t2_node)
    # Equivalent: t3_node >> t2_node

Note that runs_before modifies the other node's _upstream_nodes list, not self:

# From Node.runs_before source
def runs_before(self, other: Node):
    if self not in other._upstream_nodes:
        other._upstream_nodes.append(self)

Accessing Outputs from create_node​

When a task returns outputs, create_node attaches each output as both a named attribute and a dictionary entry on the returned Node:

@task
def t1(a: str) -> str:
    return a + " world"

@task
def t2(a: int) -> (int, str):  # two outputs
    return a + 1, str(a)

@workflow
def my_wf(a: str) -> (str, int, str):
    t1_node = create_node(t1, a=a)
    t2_node = create_node(t2, a=3)
    # Access by attribute name
    return t1_node.o0, t2_node.o0, t2_node.o1

    # Also accessible via dict:
    # t1_node.outputs["o0"]
    # t2_node.outputs["o0"], t2_node.outputs["o1"]

For single-output tasks, the output name is o0 by default. For named-tuple outputs, the names from the NamedTuple declaration are used. If a node already has an attribute matching the output name, create_node raises FlyteAssertion.

The >> Operator on Nodes​

The right-shift operator is syntactic sugar for runs_before:

t3_node >> t2_node   # same as t3_node.runs_before(t2_node)

When used on Promise or VoidPromise, the operator chains the underlying nodes:

@workflow
def wf():
    c = create_cluster(name=name)
    t = t1(a=1, b="2")
    d = delete_cluster(name=name)
    c >> t >> d  # c runs before t, t runs before d

create_node Requires Keyword Arguments​

Passing positional arguments raises FlyteAssertion:

# This fails:
create_node(t1, "hello")  # FlyteAssertion: Only keyword args are supported

# This works:
create_node(t1, a="hello")

Node.outputs Raises on Non-create_node Nodes​

The outputs property checks whether _outputs has been populated. Only nodes created via create_node() have this set. Calling node.outputs on a node created through a normal task call raises AssertionError.

Per-Node Overrides Deep Dive​

The Node.with_overrides() method mutates the node in-place and returns self. It accepts these parameters:

ParameterTypeDescription
node_namestrCustom node ID (DNS-sanitized via _dnsify)
aliasesDict[str, str]Rename output variables for the compiled workflow
resourcesResourcesCPU/memory requests and limits (preferred API)
requests / limitsResourcesDeprecated — use resources instead
timeoutint or timedeltaExecution timeout (None unsets it, 0 means no timeout)
retriesintNumber of retry attempts
interruptibleboolWhether the task can be interrupted
cachebool or CacheEnable caching with optional Cache(version=..., serialize=...)
container_imagestrOverride the container image
acceleratorBaseAcceleratorGPU/accelerator requirement
shared_memorybool or strShared memory configuration
pod_templatePodTemplateCustom pod specification

Resource Conflicts​

The resources parameter cannot be combined with limits or requests. Doing so raises ValueError:

# Raises ValueError
node.with_overrides(resources=Resources(cpu="2"), limits=Resources(cpu="1"))

# Correct
node.with_overrides(resources=Resources(cpu="2", mem="1Gi"))

Cache Override Rules​

Two forms are supported:

# Form 1: boolean True (uses default Cache settings)
node.with_overrides(cache=True)

# Form 2: Cache object with explicit version
node.with_overrides(cache=Cache(version="v1", serialize=False))

The Cache object must specify a version. Setting cache=True without cache_version enables caching with default policy. Mixing the Cache object with the deprecated cache_serialize or cache_version keyword arguments raises ValueError.

Name Sanitization​

Node IDs pass through _dnsify() to ensure DNS compliance. A custom name set via node_name is sanitized automatically.

on_failure Handlers​

When a workflow raises an exception, you can specify a handler task (or sub-workflow) to run for cleanup or notification. The handler must accept every workflow input and optionally an err: Optional[FlyteError] parameter.

Decorated Workflow with on_failure​

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

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

@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

When t1 raises, WorkflowBase.__call__ catches the exception (see workflow.py line 317):

try:
    return flyte_entity_call_handler(self, *args, **input_kwargs)
except Exception as exc:
    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

The FlyteError object is populated with the failed node's ID and the exception message. The handler receives all original workflow inputs plus err, then the exception is re-raised.

Signature Validation​

During compilation, _validate_add_on_failure_handler checks two invariants (lines 800-811):

  1. Workflow inputs must be a subset of the handler's inputs. If the handler is missing even one workflow input, FlyteFailureNodeInputMismatchException is raised.
  2. Any additional handler inputs beyond the workflow's inputs must be Optional. This enforces that the only extra data the handler can receive is optional — primarily the err parameter.
# This validates:
if (failure_node_inputs | workflow_inputs) != failure_node_inputs:
    raise FlyteFailureNodeInputMismatchException(...)
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(...)

Imperative Workflow with on_failure​

For ImperativeWorkflow, call add_on_failure_handler():

from flytekit.core.workflow import ImperativeWorkflow
from flytekit.core.python_function_task import EagerFailureHandlerTask

wb = ImperativeWorkflow(name="my.workflow.a")
in1 = wb.add_workflow_input("in1", int)
wb.add_workflow_input("in2", int)
in3 = wb.add_workflow_input("in3", int)
node = wb.add_entity(t1, a={"a": [in1, wb.inputs["in2"]], "b": [wb.inputs["in2"], in3]})
wb.add_workflow_output("from_n0t1", node.outputs["o0"])
wb.add_entity(t2)

failure_task = EagerFailureHandlerTask(name="sample-failure-task", inputs=wb.python_interface.inputs)
wb.add_on_failure_handler(failure_task)

Internally, add_on_failure_handler performs the same signature validation, creates a node via create_node, then pops it from the compilation state so it doesn't appear in the main workflow graph. The node's ID is set to the constant DEFAULT_FAILURE_NODE_ID.

ImperativeWorkflow​

ImperativeWorkflow (flytekit.core.workflow.ImperativeWorkflow) is a programmatic alternative to the @workflow decorator. Use it when you need to construct workflows dynamically rather than declaring them as function bodies.

Basic Construction​

from flytekit.core.workflow import Workflow  # alias for ImperativeWorkflow
from flytekit import task
import typing

@task
def t1(a: str) -> str:
    return a + " world"

@task
def t2():
    print("side effect")

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

# Local execution
result = wb(in1="hello")  # "hello world"

This is identical on the backend to:

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

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

API Overview​

MethodDescription
add_workflow_input(name, python_type)Declare a workflow-level input. Returns a Promise also stored in self.inputs[name].
add_entity(entity, **kwargs)Add a task, sub-workflow, or launch plan node. Returns the Node.
add_workflow_output(name, value)Declare a workflow-level output binding.
add_on_failure_handler(entity)Set a failure handler (validated, then stored separately).
add_subwf(name)Add a nested workflow.
inputsDict[str, Promise] — the declared input promises.

Unbound Input Detection​

ImperativeWorkflow tracks which declared inputs have been consumed via _unbound_inputs. If you declare an input but never pass it to any entity, flytekit raises an error early (rather than waiting for Admin's compile-time check).

Gotchas and Common Mistakes​

Workflow Body is Compile-Only​

Plain Python code inside @workflow runs only during local testing. It does not execute on the remote Flyte cluster.

# BAD: This print and conditional won't run remotely
@workflow
def bad_wf(x: int):
    print("hello")  # only during local execution
    if x > 5:       # only during local execution
        y = t1(a=x)

Use conditional() from flytekit.core.condition for runtime branching, and implement side effects in tasks.

Promises Are Not Python Values​

  • len(task_output) raises ValueError
  • range(task_output) raises ValueError
  • if task_output: raises ValueError (use task_output.is_true() or comparison)
  • and/or on promises raises ValueError (use &/| instead)

Node.outputs is Not Always Available​

Calling node.outputs on a node created through a normal task call (not create_node) raises AssertionError. Only nodes produced by create_node() have the _outputs dict populated.

create_node Accepts Only Keyword Arguments​

create_node(t1, "hello")  # FlyteAssertion
create_node(t1, a="hello")  # Correct

on_failure Handler Signature is Strictly Enforced​

The handler must accept every workflow input. Any additional parameters beyond the workflow's inputs must be Optional. A missing required parameter raises FlyteFailureNodeInputMismatchException.

With Overrides Mutates the Node​

Node.with_overrides() modifies attributes in-place and returns self. If you hold multiple references to the same Node, all references see the changes.

Cache Override Requires a Version​

node.with_overrides(cache=Cache(version="v1"))  # OK
node.with_overrides(cache=Cache())              # ValueError: must specify cache version

Node Names Are DNS-Sanitized​

Custom node names pass through _dnsify(), which ensures the name complies with Kubernetes DNS subdomain rules. The sanitized name is what flies to the platform, not necessarily the string you provided.