Conditional and dynamic workflows

Flytekit provides two primary mechanisms for introducing logic into your workflows: Conditional Branches and Dynamic Workflows. While both allow you to control execution flow, they operate at different stages of the Flyte lifecycle and have distinct constraints.

Conditional Branches

Conditional branches allow you to execute different tasks based on the values of workflow inputs or task outputs. Unlike standard Python if statements, which are evaluated at workflow registration time, Flytekit's conditional construct is evaluated at runtime by the Flyte engine.

Using the Fluent API

You define conditions using the conditional function in condition.py. This function returns a ConditionalSection that supports a fluent API: if_, elif_, then, else_, and fail.

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 my_workflow(my_input: float) -> float:
    return (
        conditional("fractions")
        .if_((my_input > 0.1) & (my_input < 1.0))
        .then(double(n=my_input))
        .elif_((my_input >= 1.0) & (my_input < 10.0))
        .then(square(n=my_input))
        .else_()
        .fail("Input out of range")
    )

Compilation vs. Execution Semantics

When you define a conditional block, Flytekit behaves differently depending on the context:

1. Compilation Mode: When the workflow is being registered, ConditionalSection captures all possible branches into a BranchNode. This allows the Flyte backend to see the entire structure of the condition before any code runs.
2. Local Execution: During local testing, LocalExecutedConditionalSection evaluates the expressions immediately. It uses ctx.execution_state.take_branch() to track which path is active and ensures only the selected task is executed.
3. Remote Execution: The Flyte engine (Propeller) evaluates the boolean expressions at runtime and schedules only the task in the matching branch.

Constraints and Limitations

Because conditions are compiled into a static graph, they have strict requirements:

• Comparison Operators: You must use bitwise operators & (AND) and | (OR) for conjunctions. Standard Python and, or, and not will not work because they return evaluated Python booleans rather than the ComparisonExpression or ConjunctionExpression objects required by the Case class.
• Unary Expressions: You cannot use if_(x) where x is a Promise. You must use an explicit comparison like if_(x == True).
• Deterministic Outputs: All branches in a conditional block must return the same type. The ConditionalSection.compute_output_vars method calculates the intersection of output variables across all branches to ensure consistency.
• No Dangling Ifs: Every conditional must end with either an else_().then(...) or an else_().fail(...).

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow depends on runtime data that cannot be determined at compilation time—for example, when the number of tasks to run depends on the size of an input list.

The @dynamic Decorator

A dynamic workflow is defined using the @dynamic decorator, which is a specialized task execution mode (PythonFunctionTask.ExecutionBehavior.DYNAMIC).

from flytekit import dynamic, task

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

@dynamic
def my_dynamic_subwf(items: list[int]) -> list[int]:
    results = []
    for i in items:
        # In a @dynamic task, you can use Python loops and logic
        # on Flyte entities.
        results.append(process_item(item=i))
    return results

How Dynamic Workflows Work

Internally, a dynamic workflow is treated as a task by the Flyte engine. However, when that task executes:

1. The function body runs locally on the worker.
2. Instead of returning data, it produces a new workflow graph (a WorkflowTemplate) based on the runtime inputs.
3. Flytekit returns this generated graph to the Flyte engine.
4. The engine then executes the generated graph as a subworkflow.

When to Use Dynamic vs. Conditional

Feature	Conditional Branch	Dynamic Workflow
Evaluation Time	Runtime (by Flyte Engine)	Runtime (by User Code)
Graph Structure	Static (all paths known at registration)	Dynamic (graph built at runtime)
Python Logic	Restricted to ComparisonExpression	Full Python (loops, recursion, etc.)
Overhead	Low (simple branch evaluation)	Higher (requires a task execution to build the graph)
Best For	Simple if/else logic on task outputs	Data-dependent parallelism (e.g., processing a list)

Use Conditional Branches when you have a fixed set of possible paths. Use Dynamic Workflows when you need to generate a variable number of tasks or complex dependencies based on runtime data. For large-scale parallel processing of identical items, consider using map_task instead of @dynamic for better performance.