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 support different types of logic.

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 or workflow inputs. Unlike standard Python if statements, which are evaluated during workflow compilation, conditional blocks are compiled into a BranchNode in the Flyte workflow graph.

Usage and Syntax

You initiate a conditional block using the conditional() function from flytekit.core.condition. This function returns a ConditionalSection that supports if_, elif_, and else_ methods.

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(should_succeed: bool) -> str:
    return (
        conditional("check-success")
        .if_(should_succeed == True)
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Supported Expressions

Flytekit conditionals do not support arbitrary Python expressions because the logic must be serializable into a Flyte BranchNode. Instead, you must use specific comparison and conjunction operators provided by Flytekit's Promise objects (defined in flytekit.core.promise):

• Comparisons: ==, !=, <, <=, >, >=
• Conjunctions: & (AND), | (OR)
• Boolean Methods: Promise.is_true(), Promise.is_false()

For example, to check if a value is within a specific range:

v = (
    conditional("range-check")
    .if_((my_input > 0.1) & (my_input < 1.0))
    .then(task_a(n=my_input))
    .else_()
    .then(task_b(n=my_input))
)

Compilation vs. Local Execution

The behavior of ConditionalSection changes depending on the context:

1. Compilation Mode: When the workflow is being compiled (e.g., for registration), ConditionalSection records all branches and their associated tasks. It uses to_branch_node to create a BranchNode that Flyte Propeller uses to route execution.
2. Local Execution: When running the workflow locally, LocalExecutedConditionalSection eagerly evaluates the expressions. It uses ctx.execution_state.take_branch() to track which path was taken and only executes the tasks in the selected branch.

Handling Errors with fail()

You can explicitly fail a workflow branch using the .fail() method on a Case. This is useful for enforcing validation logic within the workflow graph.

v = (
    conditional("validate-input")
    .if_(my_input < 0.7)
    .then(process_task(n=my_input))
    .else_()
    .fail("Input value must be less than 0.7")
)

Dynamic Workflows

Dynamic workflows, defined using the @dynamic decorator, are a hybrid between tasks and workflows. While a standard @workflow is compiled entirely before execution, a @dynamic task's body is executed at runtime to generate a new workflow plan based on actual data.

When to Use Dynamic Workflows

Use @dynamic when the structure of your workflow (the number of nodes or their dependencies) depends on the value of an input that isn't known until runtime. Common scenarios include:

• Processing a variable number of files found in a directory.
• Implementing recursive algorithms like Merge Sort.
• Building a pipeline where the number of parallel branches is determined by a previous task's output.

Implementation Details

A @dynamic function is technically a PythonFunctionTask with its execution_mode set to PythonFunctionTask.ExecutionBehavior.DYNAMIC.

from flytekit import task, dynamic
import typing

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

@dynamic
def my_dynamic_wf(n: int) -> typing.List[int]:
    results = []
    for i in range(n):
        results.append(process_item(item=i))
    return results

In the example above, the range(n) call is valid because the function body runs at execution time. In a standard @workflow, n would be a Promise object, and range(n) would raise a TypeError.

Key Differences

Feature	Conditional Section	Dynamic Workflow
Evaluation Time	Evaluated by Flyte Propeller at runtime.	Evaluated by a Flyte worker at runtime.
Graph Structure	Fixed at compile time (all branches exist).	Generated at runtime (dynamic nodes).
Python Logic	Restricted to Flyte comparison operators.	Supports full Python (loops, recursion).
Overhead	Low (simple metadata evaluation).	Higher (requires starting a task to generate the plan).

Nesting and Composition

Flytekit allows you to nest these constructs. A conditional block can return the result of a @dynamic task, and a @dynamic task can contain conditional blocks within the workflow it generates. This allows for complex, data-driven pipeline architectures while maintaining type safety and visibility in the Flyte UI.