Workflow composition, failure handlers, and nodes

Flytekit workflows are composed of nodes, each representing a task, launch plan, or sub-workflow. Nodes are created either implicitly (by calling a task inside a workflow) or explicitly using create_node(). Understanding the distinction between these approaches and how to configure nodes is essential for building robust workflows.

Task outputs and node creation

When you call a task inside a workflow, Flytekit returns a Promise object that wraps the task's outputs. This Promise serves two purposes: it enables the dual behavior between compilation (where it holds a reference to a NodeOutput) and local execution (where it holds an actual Literal value), and it provides methods like with_overrides() to configure the underlying node.

@task
def add(a: int, b: int) -> int:
    return a + b

@workflow
def my_wf(a: int, b: int) -> int:
    # This returns a Promise, not an int
    result = add(a=a, b=b)
    return result

The Promise object exposes comparison operators for conditionals, attribute access for nested structures, and with_overrides() for node configuration. During compilation, Promise.is_ready is False, and Promise.ref points to the NodeOutput. During local execution, Promise.is_ready is True, and Promise.val contains the actual value.

@workflow
def my_wf(a: int, b: int) -> int:
    result = add(a=a, b=b)
    # During compilation, this creates a ComparisonExpression
    cond = result.is_(5)
    # During local execution, this evaluates to a bool
    if result.is_ready and result.eval() == 5:
        return 0
    return result

Imperative node creation with create_node

For imperative workflows or when you need to specify dependencies between tasks without consuming their outputs, use create_node(). This function creates a Node object and, if the task has outputs, attaches them as attributes on the node.

@task
def process_data(data: str) -> (int, str):
    return len(data), data.upper()

@workflow
def imperative_wf(data: str):
    # Create a node without consuming outputs
    node = create_node(process_data, data=data)
    
    # Access outputs via node attributes (compilation) or node.outputs dict
    length = node.o0  # First output
    upper = node.o1   # Second output
    
    # Or use dictionary-style access
    length = node.outputs["o0"]
    upper = node.outputs["o1"]

The key distinction is that create_node(...).outputs is the only way to access outputs imperatively. Ordinary task calls return Promise objects that must be dereferenced by name (e.g., add(a=1, b=2).o0 for the first output).

@workflow
def declarative_wf(data: str):
    # This returns a tuple of Promises
    result = process_data(data=data)
    
    # Access outputs by attribute (named tuple style)
    length = result.o0
    upper = result.o1

Per-node overrides with with_overrides

Both Node and Promise objects support with_overrides() to configure node-level settings like timeout, retries, resources, and caching. When called on a Promise, overrides propagate to the underlying node only during compilation (when Promise.is_ready is False).

@workflow
def my_wf(a: int, b: int):
    # Configure node via Promise (compilation only)
    result = add(a=a, b=b).with_overrides(
        timeout=datetime.timedelta(minutes=5),
        retries=3,
        requests=Resources(cpu="1", memory="1Gi"),
        cache=True,
        cache_version="1.0"
    )
    
    # Configure node via Node (compilation and local execution)
    node = create_node(add, a=a, b=b).with_overrides(
        node_name="my-add-node",
        timeout=300,
        interruptible=True
    )
    result = node.o0

Valid overrides include:

• node_name: DNS-compliant name for the node
• aliases: Map of output variable names to aliases
• requests/limits: Resource specifications
• timeout: Workflow timeout (int seconds, timedelta, or None to reset)
• retries: Number of retry attempts
• interruptible: Whether the node can be interrupted
• cache: Enable caching (must specify cache_version if True)
• container_image: Override the container image
• pod_template: Override the pod template

Resource overrides must be static values; using a Promise in requests or limits raises an AssertionError.

Failure handlers with on_failure

Workflows can specify an on_failure handler that executes when the workflow fails. The handler must accept all workflow inputs plus an optional err parameter containing a FlyteError with failed_node_id and message.

@task
def notify_on_failure(err: FlyteError, workflow_input: str):
    print(f"Workflow failed with error: {err.message}")
    print(f"Failed node: {err.failed_node_id}")
    print(f"Workflow input: {workflow_input}")

@workflow(on_failure=notify_on_failure)
def my_wf(data: str) -> int:
    # This will trigger the failure handler
    raise ValueError("Something went wrong!")

The failure handler signature validation ensures:

1. The handler has an err input (optional)
2. All other handler inputs are present in the workflow inputs
3. Any additional inputs beyond workflow inputs must be optional

If the failure handler signature doesn't match, FlyteFailureNodeInputMismatchException is raised.

Gotchas and warnings

• The Node.outputs property raises AssertionError if accessed before create_node populates it. This only occurs when nodes are created imperatively without using create_node.
• with_overrides on a Promise only propagates overrides during compilation. During local execution, overrides are ignored.
• Resource overrides (requests, limits) cannot contain Promise objects. Using a promise raises AssertionError.
• create_node returns a Node during compilation but returns the raw task output (wrapped in a tuple if multiple outputs) during local execution. This dual behavior can be confusing.
• Indexing into unschematized STRUCT types or generic types without Dict[str, ...] or @dataclass annotations raises ValueError.
• The failure_node.id used in FlyteError is an empty string if the failure node hasn't been compiled yet.
• Workflow __call__ catches exceptions and invokes on_failure, then re-raises the original exception. This means the handler runs but the workflow still fails.