Conditional and dynamic workflows

Flytekit provides two primary mechanisms for controlling execution flow based on data: Conditional Workflows and Dynamic Workflows. While both allow for branching logic, they differ fundamentally in when the execution graph is constructed and how they handle input data.

Conditional Workflows

Conditional workflows allow you to define branching logic that is evaluated at runtime by the Flyte engine. Unlike standard Python if statements, which are evaluated during workflow compilation, Flytekit's conditional construct creates a BranchNode in the workflow graph.

Defining Branches

You use the conditional function to start a branch. It requires a name and supports if_, elif_, and else_ methods. Each branch must terminate with either .then() to execute a task/subworkflow or .fail() to raise an error.

from flytekit import workflow, conditional, task

@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")
    )

Expression Constraints

The expressions used in if_ and elif_ are not standard Python booleans. Because the workflow is compiled before the actual data exists, Flytekit uses ComparisonExpression and ConjunctionExpression objects.

• Supported Operators: <, <=, >, >=, ==, !=.
• Logical Operators: Use & (AND) and | (OR). Standard Python and, or, and not will not work because they attempt to evaluate the truthiness of the Promise object immediately.
• Unary Limitations: You cannot use a Promise directly as a boolean (e.g., .if_(my_input)). It must be part of a comparison.

The Case class in flytekit/core/condition.py enforces these constraints, raising an AssertionError if it receives a raw boolean or a bare Promise.

Compilation vs. Local Execution

Flytekit handles conditionals differently depending on the context:

1. Compilation/Remote: The ConditionalSection class builds an IfElseBlock. This block is serialized into the workflow definition, allowing the Flyte Propeller engine to decide which path to take at runtime without re-invoking your Python code.
2. Local Execution: The LocalExecutedConditionalSection class evaluates the expressions immediately using the actual values. It uses ctx.execution_state.take_branch() to track which path was followed, ensuring that only the tasks in the active branch are executed locally.

Dynamic Workflows

Dynamic workflows are used when the structure of the workflow (the number of tasks or the specific dependencies) depends on the value of an input. While a conditional has a fixed set of possible paths defined at compile time, a @dynamic task generates a new workflow graph at runtime.

Using the @dynamic Decorator

A dynamic workflow is defined using the @dynamic decorator. Inside the function, you can use standard Python logic—like loops and if statements—on the input values.

import typing
from flytekit import dynamic, task

@task
def t1(a: int) -> str:
    return str(a)

@dynamic
def my_dynamic_subwf(a: int) -> typing.List[str]:
    s = []
    # In a @dynamic task, you can use native Python range() and if/else
    # because the function body is executed at runtime.
    for i in range(a):
        s.append(t1(a=i))
    return s

How Dynamic Workflows Work

Internally, a @dynamic task is a PythonFunctionTask with its execution_mode set to DYNAMIC.

1. Task Execution: The Flyte engine runs the dynamic function as a normal task.
2. Graph Generation: Instead of returning data, the function returns Promise objects from tasks it calls. Flytekit captures these calls and compiles them into a WorkflowTemplate.
3. Subworkflow Execution: The engine then executes this generated template as a subworkflow.

When to Use Which

Feature	Conditional (conditional)	Dynamic (@dynamic)
Graph Structure	Fixed at compile time.	Determined at runtime.
Python Logic	Restricted to Flyte expressions (&, `	`, comparisons).
Performance	Low overhead; handled by the engine.	Higher overhead; requires a task execution to generate the graph.
Visibility	All possible branches are visible in the UI.	The graph is only visible after the dynamic task runs.

Use Conditional Workflows for simple branching based on data values. Use Dynamic Workflows when you need to parallelize over a dynamic list of items or when the workflow structure itself is complex and data-dependent.