Conditional and dynamic workflows

When to Use Conditionals

Use conditional branches when you need to route execution down one of two or more mutually exclusive paths based on a runtime value. In a Flyte workflow, this means you want Flyte Propeller to evaluate a condition and decide which branch to execute at workflow runtime.

The key distinction is this: if the decision depends on a workflow input or a task's output that only exists at runtime, you need a conditional. If you already know which branch runs at compile time, you don't need one.

Conditionals only work inside @workflow-decorated functions. Calling conditional() outside that context raises AssertionError with the message "Branches can only be invoked within a workflow context!".

Basic Syntax

The entry point is the conditional("name") function from flytekit.core.condition. Chain .if_(), .elif_(), and .else_() calls, ending each branch with .then(task_or_workflow()). The else clause is required.

from flytekit import workflow, task
from flytekit.core.condition import conditional


@task
def inc(x: int) -> int:
    return x + 1


@workflow
def example(a: int) -> int:
    return conditional("check").if_(a > 5).then(inc(x=a)).else_().then(inc(x=a))

The comparison a > 5 creates a ComparisonExpression (via Python's __gt__ operator overload on the Promise object). When Propeller evaluates this workflow, it inspects the value of a and picks the appropriate branch.

Comparison Expressions

Promise objects (the outputs of tasks and sub-workflows) support standard comparison operators:

conditional("compare").if_(a == 5).then(task1()).else_().then(task2())
conditional("compare").if_(a != 0).then(task1()).else_().then(task2())
conditional("compare").if_(a < 10).then(task1()).else_().then(task2())
conditional("compare").if_(a <= 10).then(task1()).else_().then(task2())
conditional("compare").if_(a > 0).then(task1()).else_().then(task2())
conditional("compare").if_(a >= 0).then(task1()).else_().then(task2())

For boolean values, Promise provides .is_true() and .is_false() helpers:

from flytekit import workflow, task
from flytekit.core.condition import conditional


@task
def get_flag() -> bool:
    return True


@workflow
def bool_example(flag: bool = True) -> bool:
    return conditional("flag_check").if_(flag == True).then(get_flag()).else_().then(get_flag())

Conjunction Expressions

Combine comparisons with the bitwise operators & (AND) and | (OR). Python's logical and, or, and not are not supported — they raise an AssertionError:

conditional("range").if_((a > 0) & (a < 10)).then(task1()).else_().then(task2())
conditional("multi").if_((a == 1) | (a == 2)).then(task1()).else_().then(task2())

Chain & and | to build complex expressions:

conditional("compound").if_((a > 0) & (a < 100) | (a == 999)).then(task1()).else_().then(task2())

The ComparisonExpression and ConjunctionExpression classes are defined in flytekit/core/promise.py. They implement .eval() for local execution and produce Flyte's core protobuf models for remote execution.

Elif Chains and Error Handling

Add multiple .elif_() branches before the final .else_():

from flytekit import workflow, task, conditional


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


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


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


@workflow
def grade_example(score: int) -> int:
    return (
        conditional("grade")
        .if_(score >= 90)
        .then(large(x=score))
        .elif_(score >= 50)
        .then(medium(x=score))
        .else_()
        .then(small(x=score))
    )

Use .fail() in the else branch to raise a named Error at runtime when input validation fails:

from flytekit import workflow, task, conditional


@task
def process(x: int) -> int:
    return x


@workflow
def validated_example(x: int) -> int:
    return (
        conditional("validate")
        .if_(x > 0)
        .then(process(x=x))
        .elif_(x < 0)
        .then(process(x=-x))
        .else_()
        .fail("x must be non-zero")
    )

When Propeller encounters a .fail() branch at runtime, it marks the workflow execution as failed with the given error message.

Nested Conditionals

Conditionals can appear inside .then() blocks to create nested branches:

from flytekit import workflow, task, conditional


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


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


@workflow
def nested_example(n: int) -> int:
    return (
        conditional("outer")
        .if_((n > 0) & (n < 100))
        .then(
            conditional("inner")
            .if_(n < 50)
            .then(double(n=n))
            .elif_((n >= 50) & (n < 75))
            .then(square(n=n))
            .else_()
            .fail("Value outside acceptable range")
        )
        .else_()
        .then(double(n=n))
    )

When a parent branch evaluates to false, Flytekit uses SkippedConditionalSection (via BranchEvalMode.BRANCH_SKIPPED) to prevent evaluation of expressions in child conditionals. Skipped conditionals return placeholder VoidPromise values without executing comparison logic.

Return Values and Output Types

Every branch in a conditional must return the same output type (or types, as a tuple). The ConditionalSection.compute_output_vars() method computes the intersection of output variable names across all branches. If branches return different structures, the workflow fails to compile.

from flytekit import workflow, task, conditional
import typing


@task
def left_branch(x: int) -> typing.Tuple[int, str]:
    return x, "left"


@task
def right_branch(x: int) -> typing.Tuple[int, str]:
    return x, "right"


@workflow
def tuple_output(a: int) -> typing.Tuple[int, str]:
    return conditional("pair").if_(a > 0).then(left_branch(x=a)).else_().then(right_branch(x=a))

If a branch returns None (no explicit return), the conditional resolves to VoidPromise. If all branches return None, the conditional returns None.

Compilation vs. Local Execution

The conditional() factory function selects different ConditionalSection subclasses depending on the context:

def conditional(name: str) -> ConditionalSection:
    ctx = FlyteContextManager.current_context()

    if ctx.compilation_state:
        return ConditionalSection(name)
    elif ctx.execution_state:
        if ctx.execution_state.is_local_execution():
            if ctx.execution_state.branch_eval_mode == BranchEvalMode.BRANCH_SKIPPED:
                return SkippedConditionalSection(name)
            return LocalExecutedConditionalSection(name)
    raise AssertionError("Branches can only be invoked within a workflow context!")

Compilation mode (ConditionalSection): When you run pyflyte compile, Flytekit builds a BranchNode containing an IfElseBlock from flytekit.models.core.workflow. This node is serialized to protobuf and sent to Flyte Propeller, which evaluates the condition at workflow runtime.

Local execution (LocalExecutedConditionalSection): When you run pyflyte run locally, the ComparisonExpression.eval() method actually evaluates the condition using Python's operator semantics. The selected branch's output is returned directly without constructing a BranchNode. The start_branch() method calls ctx.execution_state.take_branch() to record which branch was taken.

Skipped branches (SkippedConditionalSection): When a parent conditional's branch evaluates to false, nested conditionals receive SkippedConditionalSection. Its start_branch() skips expression evaluation and its end_branch() returns placeholder VoidPromise values.

Common Errors

Using logical operators instead of bitwise operators:

# Raises AssertionError: Logical (and/or/is/not) operations are not supported
conditional("bad").if_((a > 0) and (a < 10)).then(task1()).else_().then(task2())

Using a raw value or Promise directly in if_():

# Raises AssertionError: Flytekit does not support unary expressions
conditional("bad").if_(a).then(task1()).else_().then(task2())

Omitting the else clause:

# Raises AssertionError: A Conditional block should always end with an `else_()` clause
conditional("bad").if_(a > 0).then(task1())

Using create_node() inside a conditional:

# Raises RuntimeError: manual node creation in branch logic is disallowed

Conditionals vs. Dynamic Workflows

Conditionals and @dynamic decorated workflows serve different purposes:

Conditionals select between discrete execution paths based on a single condition. The branching structure is fixed at compile time — you know exactly which tasks can run and in what order. Propeller evaluates the condition at runtime and picks one path.
Dynamic workflows generate workflows at runtime. Use @dynamic when the number of tasks, their connections, or the structure of the graph itself depends on runtime values. For example, iterating over a list of inputs whose length is unknown until task execution.

In practice, conditionals handle "which of these N paths do I take?" and dynamic workflows handle "how many of these tasks do I need to create?" The condition module handles the former; @dynamic (defined in flytekit/core/dynamic_workflow_task.py) handles the latter.

When to Use Conditionals​

Basic Syntax​

Comparison Expressions​

Conjunction Expressions​

Elif Chains and Error Handling​

Nested Conditionals​

Return Values and Output Types​

Compilation vs. Local Execution​

Common Errors​

Conditionals vs. Dynamic Workflows​