Conditional and dynamic workflows

Flytekit provides two primary ways to handle logic that depends on runtime data: Conditional Branches and Dynamic Workflows. While both allow your code to adapt based on inputs, they differ fundamentally in how they are compiled and executed.

Conditional Branches

When you need to choose between different execution paths based on the output of a previous task, use the conditional construct. Unlike a standard Python if statement, which is evaluated at workflow compilation time, a conditional block is compiled into a BranchNode and evaluated by the Flyte engine at runtime.

Basic Usage

A conditional block starts with conditional("name") and must include at least one if_ and one else_ branch.

from flytekit import task, workflow, conditional

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

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

@workflow
def my_workflow(val: int) -> str:
    return (
        conditional("check_value")
        .if_(val > 0)
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Conditional Expressions

Conditional expressions in flytekit use specific operators because they operate on Promise objects (future values) rather than concrete Python types.

• Comparison: Use standard operators like ==, !=, <, <=, >, >=.
• Conjunction: Use & for AND and | for OR.
• Unsupported: Standard Python and, or, and not will raise an AssertionError because they cannot be overridden to produce the necessary expression trees.

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

Multi-branch and Failure

You can chain multiple conditions using .elif_() and explicitly fail a workflow using .fail().

@workflow
def complex_workflow(val: int) -> str:
    return (
        conditional("multi_check")
        .if_(val == 1)
        .then(task_one())
        .elif_(val == 2)
        .then(task_two())
        .else_()
        .fail("Value must be 1 or 2")
    )

Internal Implementation

When flytekit encounters a conditional block during workflow compilation:

1. It creates a ConditionalSection (found in condition.py).
2. Each .if_, .elif_, or .else_ call creates a Case object.
3. The .then() or .fail() calls trigger ConditionalSection.end_branch().
4. Once the else_ branch is reached (marked by last_case=True), flytekit calls to_branch_node to transform the collected cases into a BranchNode for the Flyte backend.

Dynamic Workflows

While conditional branches choose between predefined paths, Dynamic Workflows allow you to generate the workflow structure itself at runtime. Use the @dynamic decorator when the number of tasks or the specific dependencies between them depend on input data (e.g., processing a list of files where the list size is unknown until runtime).

Usage and Capabilities

A @dynamic task is run at execution time like a normal task, but its return value is a workflow that Flyte then executes. This allows you to use native Python logic like loops and standard if statements on task inputs.

from flytekit import dynamic, task

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

@dynamic
def my_dynamic_wf(items: list[int]) -> list[int]:
    results = []
    for i in items:
        # This loop is evaluated at runtime
        results.append(process_item(item=i))
    return results

Comparison: Conditional vs. Dynamic

Feature	Conditional (conditional)	Dynamic (@dynamic)
Evaluation Time	Runtime (by Flyte Engine)	Runtime (by executing the task body)
Structure	Fixed at compile time	Generated at runtime
Python Logic	Limited to &, `	`, and comparisons
Visibility	All branches visible in UI	Subworkflow generated on-the-fly
Use Case	Simple branching logic	Data-dependent parallelism (e.g. Map-Reduce)

Constraints and Performance

Dynamic workflows are powerful but come with overhead. As noted in dynamic_workflow_task.py, the generated workflow must be compiled and processed by the engine at runtime.

• Scale: Keep dynamic workflows to a reasonable number of tasks (typically under 50). For massive parallelism, prefer map_task.
• Context: A @dynamic function behaves like a @workflow during its execution phase; it cannot perform side effects like writing to a database directly; it must call tasks to perform work.

Nested Conditionals

Flytekit supports nesting conditional blocks within the .then() clause of another conditional. This is managed by SkippedConditionalSection in condition.py during local execution to ensure that branches not taken are not evaluated.

v = (
    conditional("outer")
    .if_(input_a > 0)
    .then(
        conditional("inner")
        .if_(input_b > 0)
        .then(task_both_positive())
        .else_()
        .then(task_a_positive_only())
    )
    .else_()
    .then(task_a_negative())
)

In this scenario, ConditionalSection uses FlyteContextManager to push and pop nested contexts, ensuring that nodes created within the inner branches are correctly associated with the appropriate BranchNode.