Conditional and dynamic workflows

When your workflow logic depends on the values of task outputs at runtime, you must choose between conditional branches and dynamic workflows. While both allow for non-linear execution, they differ significantly in how they are compiled and what operations they support.

Conditional Branches

Use conditional when you need to choose between a fixed set of tasks based on a simple comparison of inputs or task outputs. Conditional branches are evaluated by the Flyte engine (Propeller) at runtime, meaning the entire structure of the condition must be known at compile time.

Basic Usage

The conditional function in flytekit.core.condition allows you to build if-elif-else logic. Each branch must return a Promise (the output of a task or subworkflow) or a VoidPromise.

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: Python builtin 'bool' has no member 'is_true'

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(status: bool) -> str:
    # In a workflow, 'status' is a Promise object.
    # The Promise class provides .is_true() for boolean comparisons.
    return (
        conditional("check-status")
        .if_(status.is_true())
        .then(success_task())
        .else_()
        .then(failure_task())
    )

Supported Expressions and Constraints

Flytekit conditionals do not support arbitrary Python logic because the expressions must be serialized into Flyte's internal IfElseBlock.

1. Comparison Operators: You must use standard comparison operators: ==, !=, <, <=, >, >=. These are implemented in flytekit.core.promise.Promise to return ComparisonExpression objects.
2. Conjunctions: Use & (AND) and | (OR) instead of the Python keywords and and or.
3. Boolean Promises: When working with a boolean output from a task, use the .is_true() or .is_false() methods provided by the Promise class. Unary expressions like if_(x) are not supported.
4. Mandatory Else: Every conditional block must terminate with an .else_() clause.
5. Failure Branches: You can use .fail("error message") on a Case object to terminate a workflow branch with an error.

# Example of complex expressions and failure
# my_input is a Promise passed into the workflow
v = (
    conditional("fractions")
    .if_((my_input > 0.1) & (my_input < 1.0))
    .then(task_a(n=my_input))
    .elif_(my_input >= 1.0)
    .then(task_b(n=my_input))
    .else_()
    .fail("Input must be greater than 0.1")
)

Internal Implementation

When you call conditional(name), flytekit returns a ConditionalSection.

• During compilation, ConditionalSection captures all branches and their associated nodes. The end_branch method eventually calls to_branch_node to create a BranchNode, which encapsulates the IfElseBlock.
• During local execution, flytekit uses LocalExecutedConditionalSection. This implementation evaluates the expressions immediately using ctx.execution_state.take_branch() to determine which task to actually run, effectively short-circuiting the branches that are not taken.

Dynamic Workflows

Use the @dynamic decorator when the structure of your workflow (the number of tasks or the dependency graph) depends on runtime data. Unlike a standard @workflow, which is compiled once, a @dynamic task is executed as a task that generates a new workflow at runtime.

When to use Dynamic Workflows

Dynamic workflows are necessary when:

• You need to iterate over a list whose length is only known at runtime (e.g., processing a dynamic number of files).
• You need to use Python control flow (like for loops or if statements) directly on the values of task outputs.

from flytekit import task, dynamic, workflow
import typing

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

@dynamic
def my_dynamic_task(count: int) -> typing.List[int]:
    # In a @dynamic task, 'count' can be used in a range()
    # This is NOT allowed in a standard @workflow
    results = []
    for i in range(count):
        results.append(process_item(item=i))
    return results

@workflow
def my_wf(c: int) -> typing.List[int]:
    return my_dynamic_task(count=c)

Compilation vs. Execution Semantics

Feature	@workflow / conditional	@dynamic
Compilation	Happens once at registration time.	The "generator" task is compiled; the resulting workflow is compiled at runtime.
Python Logic	Limited to conditional expressions.	Full Python logic allowed on input values.
Node Count	Fixed at registration.	Can vary based on inputs.
Overhead	Low (handled by Propeller).	Higher (requires running a task to generate the subworkflow).

Constraints and Best Practices

• Resource Limits: Because dynamic workflows generate new nodes at runtime, they can put pressure on the Flyte engine. The flytekit.core.dynamic_workflow_task documentation recommends keeping dynamic workflows to under 50 tasks. For larger scales, use map_task.
• Promises as Values: Inside a @dynamic function, inputs are still Promise objects, but flytekit automatically resolves them so you can treat them like native Python types (e.g., using an int in range()).
• Subworkflow Modeling: On the backend, a dynamic workflow is modeled as a task. When executed, it returns a DynamicJobSpec, which Flyte Propeller then executes as a subworkflow.