Conditional and dynamic workflows

When you need to change the execution path of a workflow based on data produced by a previous task, standard Python if statements will not work because they are evaluated at compilation time. flytekit provides two primary mechanisms for runtime branching: Conditional Workflows for static graph branching and Dynamic Workflows for runtime graph generation.

Conditional Workflows

Conditional workflows allow you to define multiple execution paths within a single workflow graph. Unlike Python's native if statements, which execute during the workflow's registration phase, flytekit.conditional creates a BranchNode in the Flyte execution graph that is evaluated by the Flyte engine at runtime.

Basic Branching Logic

The conditional function in flytekit.core.condition is the entry point for creating these branches. It follows a fluent API pattern: conditional("name").if_(...).then(...).elif_(...).then(...).else_().then(...).

from flytekit import task, workflow, conditional

@task
def double(n: int) -> int:
    return n * 2

@task
def square(n: int) -> int:
    return n * n

@workflow
def my_workflow(n: int) -> int:
    return (
        conditional("compute_path")
        .if_(n > 10)
        .then(square(n=n))
        .else_()
        .then(double(n=n))
    )

Internally, conditional returns a ConditionalSection. When you call .then(), it returns the result of ConditionalSection.end_branch(). If the branch is the final else_ (marked by last_case=True), the ConditionalSection compiles the entire block into a BranchNode and adds it to the workflow's compilation_state.

Building Expressions

Conditions are built using Promise objects (the outputs of tasks). flytekit supports standard comparison operators and logical conjunctions, but it does not support Python's native and, or, and not keywords.

• Supported Comparisons: ==, !=, >, <, >=, <=
• Supported Conjunctions: & (AND), | (OR)
• Unary Constraints: You cannot use a boolean promise directly (e.g., if_(my_bool_promise)). You must compare it explicitly: if_(my_bool_promise == True).

# Valid conjunction expression
# .if_((my_input > 0.1) & (my_input < 1.0))

If you attempt to use native Python logical operators, the Case class will raise an AssertionError because those operators evaluate the Promise objects to booleans immediately during compilation rather than building a ComparisonExpression.

Nested Conditionals

You can nest conditional blocks to handle complex logic. The inner conditional must also be a complete expression returning a value to the outer then() clause.

from flytekit import task, workflow, conditional

@task
def double(n: float) -> float:
    return n * 2.0

@task
def square(n: float) -> float:
    return n * n

@workflow
def nested_workflow(my_input: float) -> float:
    return (
        conditional("fractions")
        .if_((my_input > 0.1) & (my_input < 1.0))
        .then(
            conditional("inner_fractions")
            .if_(my_input < 0.5)
            .then(double(n=my_input))
            .else_()
            .fail("Only <0.5 allowed in this branch")
        )
        .else_()
        .then(square(n=my_input))
    )

Failing a Branch

If a specific condition represents an invalid state, you can use .fail("reason") instead of .then(). This will cause the workflow execution to fail with the provided error message if that branch is taken.

Dynamic Workflows

While conditional handles branching between pre-defined tasks, Dynamic Workflows allow you to generate the workflow graph itself at runtime based on task outputs. This is useful for scenarios like processing a variable number of files or implementing recursive algorithms like merge sort.

The @dynamic Decorator

A dynamic workflow is defined using the @dynamic decorator (found in flytekit.core.dynamic_workflow_task). It acts as a hybrid between a task and a workflow:

1. Like a task: It runs at execution time on a worker.
2. Like a workflow: It returns a set of promises that the Flyte engine then executes as a subworkflow.

from flytekit import dynamic, task
from typing import List

@task
def process_item(item: int) -> int:
    return item * 2

@dynamic
def my_dynamic_subwf(count: int) -> List[int]:
    results = []
    # Native Python 'range' works here because this runs at execution time
    for i in range(count):
        results.append(process_item(item=i))
    return results

When to use Dynamic vs. Conditional

Feature	Conditional (conditional)	Dynamic (@dynamic)
Graph Structure	Static (all possible nodes known at compile time)	Dynamic (nodes generated at runtime)
Evaluation	Evaluated by Flyte Propeller (engine)	Evaluated by a Python worker
Python Logic	Limited to &, |, and comparisons	Full Python language (loops, recursion)
Overhead	Low (simple metadata check)	Higher (requires starting a pod to run the dynamic task)

Use conditional when you have a fixed set of alternative paths. Use @dynamic when the number of tasks or the specific dependencies between them cannot be determined until the workflow is already running.

Implementation Details

When a @dynamic task executes, it produces a WorkflowTemplate that is sent back to the Flyte engine. The engine then transparently executes this template as a subworkflow. In contrast, ConditionalSection works by manipulating the FlyteContextManager.current_context().compilation_state to inject a BranchNode directly into the primary workflow template during registration.