Conditional and dynamic workflows

Conditional Workflows and Dynamic Workflows

Flytekit offers two main approaches for implementing branching logic in workflows: conditional sections (statically compiled branch nodes) and dynamic workflows (runtime-constructed sub-workflows). This guide covers both, their underlying mechanics, and when to choose each.

Conditional Workflows

What conditionals are

Flyte workflows are compiled into a directed acyclic graph (DAG) before execution. Standard Python if/else statements cannot be used directly because the workflow body runs at compilation time to construct the graph, not at runtime to execute logic. Instead, flytekit provides conditional() and Workflow.create_conditional() to express branching declaratively. These get compiled into a special BranchNode in the workflow graph, which the platform (FlytePropeller) evaluates at runtime.

Basic usage

Use conditional() inside a workflow function to build an if/elif/else block. Each branch is defined with .if_(), .elif_(), or .else_(), followed by .then() which returns the output promise of the branch.

from flytekit import task, workflow, conditional

@task
def add_5(a: int) -> int:
    return a + 5

@task
def double(a: int) -> int:
    return a * 2

@workflow
def my_wf(a: int) -> int:
    return (
        conditional("branch")
        .if_(a < 10)
        .then(add_5(a=a))
        .else_()
        .then(double(a=a))
    )

The name string passed to conditional() (e.g. "branch") is arbitrary and used for identifying the node in the compiled workflow.

You can also call create_conditional on the workflow object directly:

@workflow
def my_wf(a: int) -> int:
    return (
        my_wf.create_conditional("branch")  # not valid — must be called inside the workflow function
        .if_(a < 10)
        .then(add_5(a=a))
        .else_()
        .then(double(a=a))
    )

Note: create_conditional is a method on the Workflow class and is intended to be invoked within the workflow function body, not as a chained call on the decorated function. Prefer the free-standing conditional().

Multiple branches: `elif_`

Chain as many .elif_() clauses as needed:

@workflow
def classify(a: int) -> str:
    return (
        conditional("classification")
        .if_(a < 10)
        .then(small_task())
        .elif_(a < 100)
        .then(medium_task())
        .else_()
        .then(large_task())
    )

Conditions on task outputs

You can branch on the output (a Promise) of a previously invoked task, as long as the task returns a boolean:

@task
def is_even(a: int) -> bool:
    return a % 2 == 0

@workflow
def my_wf(a: int) -> int:
    e = is_even(a=a)
    return (
        conditional("parity")
        .if_(e.is_true())
        .then(add_5(a=a))
        .else_()
        .then(double(a=a))
    )

Boolean-typed promises support .is_true() and .is_false() comparisons.

Supported expression types

Only two families of expressions are permitted in conditions:

Comparison expressions — ==, !=, <, <=, >, >=
Conjunction expressions — & (and) and | (or)

# Valid: comparison + conjunction
conditional("fractions").if_((my_input > 0.1) & (my_input < 1.0)).then(do_something())

# Invalid: Python `and` evaluates eagerly to a bool
conditional("bad").if_(my_input > 0.1 and my_input < 1.0)  # AssertionError

# Invalid: bare promise as a condition (unary expression)
conditional("bad").if_(my_input)  # AssertionError

The eager evaluation problem arises because Python's and, or, and not operators short-circuit and cannot be overloaded — my_input > 0.1 and my_input < 1.0 immediately evaluates to a plain bool before flytekit sees it. Case.__init__ detects this and raises an AssertionError:

Logical (and/or/is/not) operations are not supported. Expressions Comparison (<,<=,>,>=,==,!=) or Conjunction (&/|) are supported.

Failing a branch

.fail() allows a branch to intentionally raise an error at runtime rather than producing output:

return (
    conditional("check")
    .if_(a >= 0)
    .then(process(a=a))
    .else_()
    .fail("Input must be non-negative")
)

The error string is stored on the Case and surfaced as a runtime failure when that branch is selected.

How compilation works

When a conditional is constructed:

conditional() creates a ConditionalSection, which pushes a new compilation context via FlyteContextManager.push_context(ctx.enter_conditional_section().build()).
Each .if_()/.elif_()/.else_() call creates a Case via Condition._if(), Condition.elif_(), or Condition.else_() (the latter passes expr=None and last_case=True), and ConditionalSection.start_branch() appends it to the section's case list.
.then(promise) records the branch's output promise on the Case and calls ConditionalSection.end_branch().
When the final branch's end_branch() runs, to_branch_node() compiles the accumulated cases into a BranchNode wrapping a IfElseBlock model, and a Node is created with bindings for any unresolved (not-yet-ready) promises from upstream nodes.
compute_output_vars() computes the branch node's outputs by intersecting the output variable names across all cases — so all branches must produce outputs with the same names/types. If any branch is void, the entire conditional produces a VoidPromise.

How local execution works

When you run a workflow locally (e.g. my_wf(a=5)), flytekit uses LocalExecutedConditionalSection, a subclass of ConditionalSection:

In start_branch(), each case expression is eagerly evaluated via expr.eval(). The first case whose expression evaluates truthy (or which is the else_ case) triggers ctx.execution_state.take_branch() and is recorded as the selected case.
Subsequent cases are still registered (for graph bookkeeping) but not executed.
In end_branch(), the selected case's promise is returned, or ValueError is raised if the branch used .fail().

Nested conditionals

If a conditional appears inside a branch of an outer conditional that was not selected during local execution, flytekit uses SkippedConditionalSection. Its docstring explains:

This ConditionalSection is used for nested conditionals, when the branch has been evaluated to false. This ensures that the branch is not evaluated and thus the local tasks are not executed.

Instead of evaluating or executing anything, SkippedConditionalSection.end_branch() returns promises with None values (or a VoidPromise), so untaken inner branches are skipped without raising errors about missing values.

Key constraint: output consistency

Because compute_output_vars() intersects output names across branches, all .then() branches of a conditional must return values that can be unified into a common set of outputs. If one branch returns nothing, the whole conditional returns VoidPromise and cannot be used as a value. If branches return differently-shaped tuples, the intersection may drop variables, changing the workflow's effective output.

Dynamic Workflows

What dynamic workflows are

Conditional sections express static, fixed-shape branching. A dynamic workflow is for cases where the shape of the downstream graph isn't known until runtime — for example, fanning out over a list whose length is only known after a previous task completes, or choosing between tasks that depend on runtime data types.

A dynamic workflow is declared with the @dynamic decorator. The decorated function looks like a task — it receives concrete (resolved) values at runtime and can invoke tasks, sub-workflows, and even conditionals — but its return value is a WorkflowExecution object representing a newly constructed sub-workflow graph that the platform then executes.

from typing import List
from flytekit import task, dynamic, workflow

@task
def square(x: int) -> int:
    return x * x

@task
def sum_values(xs: List[int]) -> int:
    return sum(xs)

@dynamic
def fan_out(xs: List[int]) -> int:
    results = []
    for x in xs:  # length known only at runtime
        results.append(square(x=x))
    return sum_values(xs=results)

@workflow
def my_wf(xs: List[int]) -> int:
    return fan_out(xs=xs)

Python control flow (for, if) is legal inside a dynamic workflow because the function actually runs at runtime — the loop unrolls into a concrete sub-DAG at that moment.

Conditionals vs. dynamic workflows: when to use which

Situation	Use
Branch choice depends on a value known at workflow-construction time or a boolean/comparison on promises	`conditional()`
Number of parallel branches/nodes depends on runtime data (e.g. list length)	`@dynamic`
Branch bodies are fixed, but which branch runs is decided at runtime	`conditional()` — branches are compiled into the graph but only the selected one executes
Complex Python logic (loops, arbitrary branching) over runtime values	`@dynamic`

Execution semantics differences

Conditionals compile into a single BranchNode whose branches are evaluated by the Flyte platform at runtime. Branch bodies are eagerly compiled (task nodes exist in the graph), but only the selected branch runs.
Dynamic workflows execute the decorated function at runtime to generate a new workflow, which is then compiled and executed as a nested sub-workflow. This adds one level of workflow nesting and a small runtime overhead for graph construction, but grants full Python expressiveness inside.

Combining the two

You can use conditional() inside a @dynamic workflow to combine both mechanisms:

@dynamic
def fan_out(xs: List[int]) -> int:
    results = []
    for x in xs:
        results.append(
            conditional("parity")
            .if_(x % 2 == 0)   # note: x is a runtime int here
            .then(square(x=x))
            .else_()
            .then(double(x=x))
        )
    return sum_values(xs=results)

Inside a dynamic workflow, values like x are concrete Python values, so comparisons like x % 2 == 0 produce real booleans. However, when conditionals inside dynamic workflows operate on promises returned by tasks invoked within the dynamic function, the same Case validation rules apply (only comparison/conjunction expressions, no bare promises, no eager and/or).

Conditional Workflows and Dynamic Workflows​

Conditional Workflows​

What conditionals are​

Basic usage​

Multiple branches: elif_​

Conditions on task outputs​

Supported expression types​

Failing a branch​

How compilation works​

How local execution works​

Nested conditionals​

Key constraint: output consistency​

Dynamic Workflows​

What dynamic workflows are​

Conditionals vs. dynamic workflows: when to use which​

Execution semantics differences​

Combining the two​