Conditional and dynamic workflows

Flytekit provides two primary mechanisms for introducing non-linear logic into your pipelines: Conditional Sections and Dynamic Workflows. While both allow for branching and decision-making, they operate at different stages of the Flyte lifecycle and have distinct constraints.

Conditional Sections

Conditional sections allow you to define branching logic that is evaluated by the Flyte engine at runtime based on the outputs of previous tasks. Unlike standard Python if statements, which are evaluated during workflow compilation, conditional blocks are compiled into a BranchNode in the Flyte workflow graph.

Defining Branches

You use the conditional() function from the flytekit package to start a branch. It follows a builder pattern using .if_(), .elif_(), .else_(), and .then().

from flytekit import task, workflow, conditional

@task
def success_task() -> str:
    return "Success"

@task
def failure_task() -> str:
    return "Failure"

@workflow
def my_conditional_wf(val: int) -> str:
    return (
        conditional("value_check")
        .if_(val > 10)
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Internal Implementation

When you call conditional(name), flytekit creates a ConditionalSection object. This object manages a list of Case objects, each representing a branch.

1. if_ / elif_: These methods create a Case and register it with the ConditionalSection via start_branch.
2. then: This method associates a task or subworkflow output (a Promise) with the current Case and triggers end_branch.
3. Compilation: In ConditionalSection.end_branch, if it is the last case (the else_ block), flytekit converts the entire section into a BranchNode using to_branch_node. This node is then added to the workflow's compilation_state.

Constraints and Expressions

Because conditional blocks are compiled into a static graph, you cannot use standard Python boolean logic (like if x:) or unary promises (like if_(my_promise)). Flytekit requires explicit comparison or conjunction expressions.

The Case class (found in flytekit/core/condition.py) enforces these rules:

• Supported Operators: Comparison (<, <=, >, >=, ==, !=) and Conjunction (&, |).
• Unsupported: Logical and, or, is, not are not supported because they evaluate the Python object itself rather than building a Flyte expression.
• Unary Promises: You cannot pass a raw Promise to if_(). You must compare it to a value, e.g., my_promise.is_true() or my_promise == True.

# Valid Conjunction
# .if_((val > 0) & (val < 10))

# Invalid: This will raise an AssertionError in the Case constructor
# .if_(val > 0 and val < 10) 

Dynamic Workflows

Dynamic workflows, defined with the @dynamic decorator, are used when the structure of the workflow depends on data that is only available at runtime. A dynamic workflow is essentially a task that, when executed, returns a new workflow graph (a subworkflow) to the engine.

When to Use Dynamic Workflows

Use @dynamic when you need to:

1. Loop over data: If the number of tasks depends on the size of an input list.
2. Access Promise values: Unlike a standard @workflow, a @dynamic task allows you to treat inputs as actual Python values (e.g., using them in a range() call).

from flytekit import task, dynamic, workflow
import typing

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

@dynamic
def my_dynamic_subwf(items: typing.List[int]) -> typing.List[int]:
    # items is a real list here, not a Promise
    results = []
    for i in items:
        results.append(process_item(item=i))
    return results

@workflow
def main_wf(input_list: typing.List[int]) -> typing.List[int]:
    return my_dynamic_subwf(items=input_list)

Execution Semantics

A @dynamic function is a hybrid:

• At Compilation: It is treated as a single task node in the parent workflow.
• At Runtime: The function body executes. Instead of returning data, it returns a set of Promise objects. Flytekit captures these calls, builds a subworkflow, and submits it back to the Flyte engine for execution.

This is implemented in flytekit/core/dynamic_workflow_task.py by setting the execution_mode to PythonFunctionTask.ExecutionBehavior.DYNAMIC.

Comparison: Conditional vs. Dynamic

Feature	Conditional Section	Dynamic Workflow
Evaluation Time	Evaluated by the engine at runtime.	Evaluated by a worker at runtime.
Graph Visibility	The entire BranchNode is visible in the static graph.	The internal graph is hidden until the dynamic task runs.
Python Logic	Limited to Flyte expressions (&, |, ==).	Full Python logic (loops, if statements, range).
Overhead	Low; handled by the Flyte engine.	Higher; requires a task execution to generate the subworkflow.

Nesting and Errors

You can nest conditional blocks inside each other or inside @dynamic tasks. If a branch should result in a failure, use the .fail() method on a Case instead of .then().

from flytekit import task, workflow, conditional

@task
def success_task() -> str:
    return "Success"

@workflow
def fail_wf(val: int) -> str:
    return (
        conditional("check")
        .if_(val < 0)
        .fail("Value cannot be negative")
        .else_()
        .then(success_task())
    )

In flytekit/core/condition.py, the fail method records the error string in the Case object, which is then compiled into the IfElseBlock of the BranchNode.