Workflows and workflow composition

By the end of this tutorial, you will have built the same computation in both of flytekit's workflow styles, run it locally with native Python values, composed an imperative workflow from a subworkflow and a launch plan, and inspected the serialized nodes, outputs, metadata, and failure node.

Prerequisites

• A Python environment in which flytekit is importable.
• A task created with flytekit.core.task.task.
• Python type annotations for workflow and task inputs and outputs.
• For serialization, the SerializationSettings and image configuration used by flytekit's tests. Local calls do not require those settings.

1. Start with tasks and a function workflow

Define two tasks, then express their data flow with @workflow:

import typing

from flytekit.core.task import task
from flytekit.core.workflow import workflow


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


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


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


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


assert my_workflow(in1="hello").from_n0t1 == "hello world"

PythonFunctionWorkflow is the object created for my_workflow. Its constructor derives the workflow name and native interface from the Python function. During compile, it creates input Promise objects, evaluates the function under a compilation context, collects the task calls as Node objects, and turns the returned value into output bindings. The compiled flag prevents that compilation from being repeated for the same workflow object.

The @workflow docstring in flytekit/core/workflow.py makes an important distinction: the function body is evaluated at serialization time to discover the graph, and is not evaluated again by Flyte when the workflow runs. The print in this example belongs to the t2 task; it is not a separate runtime workflow node created by putting an ordinary side effect directly in the workflow body.

The call above is local execution. WorkflowBase.local_execute, inherited by PythonFunctionWorkflow, translates native input values into literal-backed Promise objects, calls execute, validates the returned shape against the workflow interface, and repackages the results under the workflow's output names.

Function-workflow output shape

PythonFunctionWorkflow.compile binds outputs according to the declared interface. Multiple outputs must be returned as a tuple of the matching length. A one-output NamedTuple has special handling so that its named field can become the workflow output name. The following multiple-output workflow is taken from flytekit's workflow tests:

import typing

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


@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

Here the first task produces two promises, x and y; the second task consumes x; and the returned tuple supplies the two workflow outputs. If a workflow declares multiple outputs but returns one value, or returns a tuple with the wrong length, compilation raises an assertion or value error rather than producing a mismatched interface. A conditional output must also be completed with an else_() clause before it can be bound.

2. Understand workflow metadata and failure policy

The workflow-level failure policy and the defaults inherited by workflow nodes are separate metadata layers:

from flytekit.core.workflow import (
    WorkflowFailurePolicy,
    WorkflowMetadata,
    WorkflowMetadataDefaults,
)


metadata = WorkflowMetadata(on_failure=WorkflowFailurePolicy.FAIL_IMMEDIATELY)
defaults = WorkflowMetadataDefaults(interruptible=False)

WorkflowFailurePolicy is an enum with two values:

• FAIL_IMMEDIATELY makes the workflow enter a failed state when a component node fails.
• FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows remaining runnable nodes to run after a component node fails.

WorkflowMetadata.__post_init__ accepts only those two enum values. Passing the serialized integer directly is invalid, even though to_flyte_model() maps FAIL_IMMEDIATELY to model value 0 and FAIL_AFTER_EXECUTABLE_NODES_COMPLETE to model value 1:

import pytest

from flytekit.core.workflow import WorkflowFailurePolicy, WorkflowMetadata
from flytekit.exceptions.user import FlyteValidationException


with pytest.raises(FlyteValidationException):
    WorkflowMetadata(on_failure=0)

assert WorkflowMetadata(
    on_failure=WorkflowFailurePolicy.FAIL_IMMEDIATELY
).on_failure == WorkflowFailurePolicy.FAIL_IMMEDIATELY

WorkflowMetadataDefaults currently contains the interruptible default. It validates exact boolean values, so 3, 1, and other truthy non-booleans are rejected. WorkflowBase.construct_node_metadata() reads this value when constructing node metadata. The class docstring distinguishes this inherited default from WorkflowMetadata, which describes the workflow itself.

The decorator and imperative constructor expose these settings directly. The function workflow above sets interruptible=True and FAIL_AFTER_EXECUTABLE_NODES_COMPLETE; an ImperativeWorkflow accepts the corresponding failure_policy and interruptible constructor arguments, defaulting to immediate failure and non-interruptible nodes.

3. Build the same graph imperatively

ImperativeWorkflow is the programmatic WorkflowBase implementation. Declare the workflow input, add entities, and explicitly bind the workflow output:

from flytekit.core.task import task
from flytekit.core.workflow import ImperativeWorkflow


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


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


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_n0t1", node.outputs["o0"])

assert wb(in1="hello") == "hello world"

WorkflowBase is the common storage and execution boundary for both styles. It holds the workflow name, native and typed interfaces, input promises, Node objects, output bindings, optional failure handler, documentation, and default options. ImperativeWorkflow keeps an explicit CompilationState; each add_entity call adds a node to that state. wb.inputs["in1"] is a Promise whose referenced node is flytekit's global start node, and node.outputs["o0"] is the promise used to bind the task's result to the workflow output.

add_workflow_input also records the new promise as unbound. Adding an entity consumes any input promises passed to it. Consequently, an imperative workflow is ready only when it has at least one node and no declared input remains unbound. Duplicate input and output names raise FlyteValidationException.

Imperative execution is intentionally ordered. ImperativeWorkflow.execute walks compilation_state.nodes in the order in which they were added, resolves each node's bindings from the outputs already produced, invokes the entity, and stores its outputs for later nodes. Add entities in dependency/topological order; the imperative executor does not perform a separate dependency schedule.

Collection and map bindings

add_entity recursively finds promises inside lists and dictionaries, so a task input can be assembled from several workflow inputs:

import typing

from flytekit.core.task import task
from flytekit.core.workflow import ImperativeWorkflow


@task
def t1(a: typing.Dict[str, typing.List[int]]) -> typing.Dict[str, int]:
    return {k: sum(v) for k, v in a.items()}


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

assert wb(in1=3, in2=4, in3=5) == {"a": 7, "b": 9}

The same recursive binding logic handles a list input, as in the test task that sums two imperative workflow inputs:

import typing

from flytekit.core.task import task
from flytekit.core.workflow import ImperativeWorkflow


@task
def sum_values(a: typing.List[int]) -> int:
    return sum(a)


wb = ImperativeWorkflow(name="my.workflow.a")
wb.add_workflow_input("in1", int)
wb.add_workflow_input("in2", int)
node = wb.add_entity(sum_values, a=[wb.inputs["in1"], wb.inputs["in2"]])
wb.add_workflow_output("from_n0t1", node.outputs["o0"])

assert wb(in1=3, in2=4) == 7

When the workflow output itself is represented by a list or dictionary of promises, add_workflow_output cannot infer the container's Python type from one promise. Supply its python_type explicitly in that case, for example by passing the container type as the third argument.

4. Compose subworkflows and launch plans

Imperative workflows accept a WorkflowBase subworkflow and a LaunchPlan as entities. The resulting node refers to the appropriate workflow or launch-plan object rather than becoming a copy of the child graph:

import typing
from collections import OrderedDict

import flytekit.configuration
from flytekit.configuration import Image, ImageConfig
from flytekit.core.launch_plan import LaunchPlan
from flytekit.core.task import task
from flytekit.core.workflow import ImperativeWorkflow
from flytekit.tools.translator import get_serializable


default_img = Image(name="default", fqn="test", tag="tag")
serialization_settings = flytekit.configuration.SerializationSettings(
    project="project",
    domain="domain",
    version="version",
    env=None,
    image_config=ImageConfig(default_image=default_img, images=[default_img]),
)


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


child = ImperativeWorkflow(name="my_workflow")
child.add_workflow_input("in1", str)
child_node = child.add_entity(t1, a=child.inputs["in1"])
child.add_workflow_output("from_n0t1", child_node.outputs["o0"])

launch_plan = LaunchPlan.create("test_wb", child)

parent_from_workflow = ImperativeWorkflow(name="parent.imperative")
p_in1 = parent_from_workflow.add_workflow_input("p_in1", str)
p_node0 = parent_from_workflow.add_subwf(child, in1=p_in1)
parent_from_workflow.add_workflow_output("parent_wf_output", p_node0.from_n0t1, str)

parent_spec = get_serializable(
    OrderedDict(), serialization_settings, parent_from_workflow
)
assert parent_spec.template.nodes[0].workflow_node.sub_workflow_ref.name == "my_workflow"

parent_from_launch_plan = ImperativeWorkflow(name="parent.imperative")
p_in1 = parent_from_launch_plan.add_workflow_input("p_in1", str)
p_node0 = parent_from_launch_plan.add_launch_plan(launch_plan, in1=p_in1)
parent_from_launch_plan.add_workflow_output(
    "parent_wf_output", p_node0.from_n0t1, str
)

launch_plan_spec = get_serializable(
    OrderedDict(), serialization_settings, parent_from_launch_plan
)
assert launch_plan_spec.template.nodes[0].workflow_node.launchplan_ref.name == "test_wb"

add_subwf, add_launch_plan, and add_task are convenience methods that delegate to add_entity. The node's output promise can be used exactly like a task output promise, which is why p_node0.from_n0t1 can be bound to the parent's output.

5. Add a failure handler separately from the failure policy

A workflow failure policy determines what happens after a component node fails. An on_failure handler is a separate task or workflow used to handle failure. For an imperative workflow, add it with add_on_failure_handler:

from flytekit.core.python_function_task import EagerFailureHandlerTask


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

The imperative implementation creates the handler node using the workflow inputs, then removes it from the normal node list and assigns the reserved default failure-node ID. Therefore, the handler is not another ordinary node scheduled with the main task nodes. The corresponding serialization check in flytekit's tests is:

from collections import OrderedDict

from flytekit.models.admin.workflow import WorkflowSpec
from flytekit.tools.translator import get_serializable


wf_spec: WorkflowSpec = get_serializable(
    OrderedDict(), serialization_settings, wb
)
assert wf_spec.template.failure_node is not None
assert wf_spec.template.failure_node.id == "efn"

Failure-handler interfaces are validated. The handler must accept the workflow inputs; any additional handler inputs must be optional. For a function workflow, PythonFunctionWorkflow._validate_add_on_failure_handler compiles the handler in a separate compilation context and requires exactly one resulting task or workflow node.

At invocation time, WorkflowBase.__call__ also catches an exception from the entity call. If on_failure is configured, it invokes that handler; when the handler declares an input named err, flytekit supplies a FlyteError containing the failed node ID and exception message. This call-time handler behavior is distinct from the serialized WorkflowMetadata.on_failure scheduling policy.

6. Inspect the compiled graph and serialized metadata

Use get_serializable with the same settings shape used by flytekit's tests:

import typing
from collections import OrderedDict

import flytekit.configuration
from flytekit.configuration import Image, ImageConfig
from flytekit.core.task import task
from flytekit.core.workflow import WorkflowFailurePolicy, workflow
from flytekit.tools.translator import get_serializable


default_img = Image(name="default", fqn="test", tag="tag")
serialization_settings = flytekit.configuration.SerializationSettings(
    project="project",
    domain="domain",
    version="version",
    env=None,
    image_config=ImageConfig(default_image=default_img, images=[default_img]),
)


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


@workflow(
    interruptible=True,
    failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE,
)
def configured_workflow(a: int) -> int:
    return add_two(a=a)


wf_spec = get_serializable(
    OrderedDict(), serialization_settings, configured_workflow
)
assert wf_spec.template.metadata_defaults.interruptible
assert wf_spec.template.metadata.on_failure == 1
assert len(wf_spec.template.nodes) == 1
assert len(wf_spec.template.outputs) == 1

The serializer sees the nodes and output bindings collected by PythonFunctionWorkflow.compile, and it sees the two metadata layers separately: metadata_defaults.interruptible comes from WorkflowMetadataDefaults, while metadata.on_failure comes from WorkflowMetadata.

For an imperative workflow, inspect wb.nodes, wb.python_interface, and wb.output_bindings after construction, or serialize it as in the composition example. WorkflowBase exposes the native python_interface, translated typed interface, nodes, output_bindings, on_failure, and failure_node properties so the graph and its interface can be examined before registration.

7. See where imperative workflows are reused internally

The imperative API is also the representation used by flytekit's eager-task integration. PythonFunctionTask.get_as_workflow declares every task input on an ImperativeWorkflow, adds the task as one node, binds every task output, and attaches an eager cleanup handler:

from flytekit.core.workflow import ImperativeWorkflow


cleanup = EagerFailureHandlerTask(
    name=f"{self.name}-cleanup",
    container_image=self.container_image,
    inputs=self.python_interface.inputs,
)
wb = ImperativeWorkflow(name=self.name)

input_kwargs = {}
for input_name, input_python_type in self.python_interface.inputs.items():
    wb.add_workflow_input(input_name, input_python_type)
    input_kwargs[input_name] = wb.inputs[input_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])

wb.add_on_failure_handler(cleanup)

This is the same graph-building sequence you used manually: inputs become promises, the entity becomes a node, node outputs become workflow outputs, and cleanup is kept as a failure node rather than a normal node.

Complete result and next steps

You now have two equivalent ways to express the t1/t2 graph:

• PythonFunctionWorkflow evaluates the decorated function during compilation and infers nodes and output bindings from task calls and the returned value.
• ImperativeWorkflow records each node and output binding through add_entity and add_workflow_output.

Both use WorkflowBase for interfaces, local execution, metadata, failure handling, and serialization. The imperative call wb(in1="hello") returns "hello world"; the nested list/map example returns {"a": 7, "b": 9}; and serialized parent workflows expose either sub_workflow_ref or launchplan_ref in their workflow nodes.

For further composition, use conditional sections inside a function workflow, continue using add_subwf or add_launch_plan for explicit parent graphs, and serialize with get_serializable to verify node references, output bindings, metadata, and failure nodes before deployment.