Conditional and dynamic workflows

Flytekit provides two primary mechanisms for branching logic: static conditional branches and dynamic workflows. While both allow you to control execution flow based on data, they differ fundamentally in when they are evaluated and how they affect the workflow graph.

Static Conditional Branches

Use conditional() when you want to define a fixed set of execution paths that are determined at compile time. A conditional block results in a single BranchNode in the Flyte workflow graph. The specific path taken is decided by the Flyte engine at runtime based on the values of workflow inputs or task outputs.

Defining Conditions

The conditional factory in condition.py provides a fluent API for building these branches. Every conditional block must start with .if_(), can have zero or more .elif_() clauses, and must terminate with either an .else_().then(...) or an .else_().fail(...).

from flytekit import task, workflow, conditional

@task
def add_five(a: int) -> int:
    return a + 5

@workflow
def my_conditional_wf(a: int) -> int:
    return (
        conditional("check-value")
        .if_(a == 5)
        .then(add_five(a=a))
        .elif_(a > 10)
        .then(add_five(a=10))
        .else_()
        .fail("Value must be 5 or > 10")
    )

Expression Syntax and Constraints

Flytekit conditions are not standard Python if statements. Because they must be serialized into a workflow specification, they use specific operators and types:

• Comparison Operators: Use standard comparisons like ==, !=, <, <=, >, >=.
• Conjunctions: Use the bitwise operators & (AND) and | (OR) to combine expressions.
• No Python Built-ins: You cannot use Python's and, or, not, or is keywords. These will raise an AssertionError in the Case class because they evaluate to Python booleans immediately rather than creating the necessary ComparisonExpression or ConjunctionExpression objects.
• No Bare Promises: You cannot pass a raw Promise (the output of a task) directly to .if_(promise). You must compare it to a value, e.g., .if_(promise == True) or use .if_(promise.is_true()).

Internal Compilation and Output Merging

When flytekit compiles a ConditionalSection, it performs several validation and transformation steps:

1. Context Management: ConditionalSection.__init__ pushes a new context via FlyteContextManager. This ensures that any tasks called within a .then() block are correctly associated with that specific branch.
2. Output Intersection: The compute_output_vars method calculates the intersection of output variables across all branches. If one branch returns an int and another returns nothing (VoidPromise), the entire conditional block is treated as having no output.
3. Node Creation: Once .else_() is called, end_branch triggers the creation of a Node containing a BranchNode. This node encapsulates the IfElseBlock model, which includes the conditions and the corresponding sub-nodes for each branch.
4. Binding Deduplication: The merge_promises helper in condition.py ensures that all inputs required by the various conditions are uniquely bound to the BranchNode.

Dynamic Workflows

Use the @dynamic decorator when the structure of your workflow depends on runtime data that cannot be expressed as a static branch. For example, if you need to run a task for every element in a list whose length is only known at runtime, a static conditional is insufficient.

A dynamic workflow is actually a specialized task. When it runs, it executes its Python body to generate a new workflow (a DynamicJobSpec) which is then executed by the Flyte engine.

from flytekit import dynamic, task
import typing

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

@dynamic
def my_dynamic_wf(items: typing.List[int]) -> typing.List[int]:
    results = []
    for i in items:
        # Native Python control flow (for loops, if statements) 
        # is allowed here because this runs at execution time.
        results.append(process_item(item=i))
    return results

Key Differences from Static Workflows

Feature	Static Conditional (conditional)	Dynamic Workflow (@dynamic)
Evaluation Time	Compile time (graph is fixed)	Execution time (graph is generated)
Control Flow	Fluent API (.if_().then())	Native Python (if, for, while)
Graph Visibility	Entire branch structure is visible in UI	Sub-graph is only visible after dynamic task runs
Overhead	Low (single node evaluation)	Higher (requires running a task to generate the graph)

Local Execution Behavior

Flytekit handles local execution of these constructs differently to simulate remote behavior:

• Conditionals: LocalExecutedConditionalSection evaluates the expressions against local values. It calls ctx.execution_state.take_branch() for the first matching case and skips the execution of tasks in all other branches. The promise.py call handler detects this BRANCH_SKIPPED state and returns null-valued promises instead of executing the tasks.
• Dynamic Workflows: The function body runs locally. If it is part of a larger local workflow execution, it returns the final results directly. During "hosted" dynamic execution (compiling the dynamic spec), it uses a CompilationState with a d prefix to track the generated nodes.

Nested Conditionals

Conditionals can be nested within other conditionals. In local execution, if an outer branch is skipped, flytekit uses SkippedConditionalSection for any nested blocks. This ensures that the internal structure is still processed (to maintain context stacks) without actually evaluating expressions or running tasks.

v = (
    conditional("outer")
    .if_(a > 0)
    .then(
        conditional("inner")
        .if_(a < 10)
        .then(task_a())
        .else_()
        .then(task_b())
    )
    .else_()
    .then(task_c())
)

In this scenario, if a <= 0, the SkippedConditionalSection ensures that task_a and task_b are never invoked, and the inner conditional immediately returns a VoidPromise or null-valued promises to the outer block.