Conditional and dynamic workflows

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

Conditional Sections

Conditional sections allow you to define branching logic that is evaluated by the Flyte engine at runtime. Because these branches are defined within a @workflow, the Flyte compiler must be able to see all possible execution paths upfront.

Defining Branches

You create a conditional block using the conditional function from flytekit. This returns a ConditionalSection that supports if_, elif_, and else_ methods. Every conditional block must terminate with either an else_() or a fail() call.

from flytekit import workflow, conditional, task

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

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

@workflow
def branching_wf(a: int) -> int:
    return (
        conditional("value-check")
        .if_(a > 10)
        .then(double(x=a))
        .elif_(a < 0)
        .then(add_five(x=a))
        .else_()
        .then(a)
    )

Supported Expressions

Flytekit does not support standard Python logical operators (and, or, not) or truthiness checks (e.g., if a:) within a workflow because the values are Promise objects, not actual data, during compilation. Instead, use the following:

• Comparisons: Use standard operators like ==, !=, <, <=, >, >=.
• Conjunctions: Use & (AND) and | (OR) for combining expressions.
• Boolean Methods: For boolean inputs or outputs (which are Promise objects in the workflow context), use .is_true() or .is_false().

from flytekit import workflow, conditional, task

@task
def get_val(a: int) -> int:
    return a

@workflow
def logic_wf(a: int, b: bool) -> int:
    # b is a Promise[bool] here, which supports .is_true()
    return (
        conditional("complex-logic")
        .if_((a > 0) & (b == True))
        .then(get_val(a=a))
        .else_()
        .fail("Condition not met")
    )

Internal Implementation

When you call conditional("name"), flytekit initializes a ConditionalSection. During workflow compilation, this section tracks every Case (branch) you define.

1. Context Management: ConditionalSection uses FlyteContextManager.push_context to enter a conditional state.
2. Node Creation: Each branch's then() call captures the resulting Promise.
3. Branch Resolution: The end_branch() method in ConditionalSection calculates the intersection of output variables across all branches to ensure the conditional block returns a consistent interface.
4. Backend Representation: The entire block is compiled into a BranchNode, which contains an IfElseBlock for the Flyte engine to execute.

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow depends on the actual values of the data, which are only known at runtime. Unlike standard workflows, a @dynamic function can iterate over lists, use Python if statements, and inspect the contents of its inputs.

Usage Scenario

Use @dynamic when you need to generate a variable number of tasks based on an input, such as processing every file in a directory or performing a map-reduce operation where the number of shards is determined by the data size.

from flytekit import dynamic, task
import typing

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

@dynamic
def dynamic_wf(count: int) -> typing.List[int]:
    results = []
    # Standard Python loops and range() are allowed here
    for i in range(count):
        results.append(process_item(x=i))
    return results

Compilation vs. Execution Semantics

The behavior of @dynamic is a hybrid between a task and a workflow:

• At Workflow Compilation: The dynamic task is treated as a single, opaque node. The Flyte compiler does not know what tasks it will eventually run.
• At Runtime: The dynamic task function executes. Instead of returning data directly, it returns a "spec" for a subworkflow. Flytekit captures the tasks called inside the function and sends this generated workflow back to the Flyte engine to be executed as a subworkflow.

Key Differences

Feature	Conditional Section	Dynamic Workflow
Evaluation Time	Flyte Engine (Runtime)	User Code (Runtime)
Python Logic	Restricted (use &, `	, is_true`)
Visibility	All branches visible in UI upfront	Subworkflow generated on-the-fly
Data Access	Cannot "peek" at data values	Can inspect and use input values
Use Case	Simple branching based on task outputs	Data-dependent structure/looping

Nested Conditionals

Conditionals can be nested within each other. Flytekit manages this by tracking the ConditionalSection stack. If a branch is skipped during local execution, SkippedConditionalSection ensures that the tasks within that branch are not executed, preventing unnecessary local computation.

@workflow
def nested_wf(a: int, b: int) -> int:
    return (
        conditional("outer")
        .if_(a > 0)
        .then(
            conditional("inner")
            .if_(b > 0)
            .then(add_five(x=a))
            .else_()
            .then(double(x=a))
        )
        .else_()
        .then(a)
    )