Conditional and dynamic workflows

When to use conditional branches vs. dynamic workflows

Flytekit gives you two ways to make your workflow logic depend on values that are only known at runtime: conditional branches and dynamic workflows. Although they both let you express non-linear execution paths, they differ fundamentally in when code runs and how the execution graph is constructed.

• Conditional branches (conditional() / .if_() / .then()) select one static sub-graph at runtime. Every branch is already compiled into the workflow DAG before execution starts; the runtime engine simply chooses which sub-graph to activate. Use conditionals when the number of branches is small and fixed at authoring time.
• Dynamic workflows (@dynamic) let a task body run at execution time and emit a brand-new subworkflow dynamically. The subworkflow's structure (number of nodes, their dependencies) is not known until the dynamic function executes. Use dynamic workflows when the graph itself must be data-dependent — for example, iterating over an input to create one task invocation per element.

This section covers both APIs, their semantics at compilation and execution time, and the constraints that govern their use.

Conditional branches API

Entry point: conditional()

The conditional() function in flytekit.core.condition (line 478) is the factory that starts a conditional branch chain. It takes a single name: str argument and returns a ConditionalSection instance:

from flytekit import conditional

conditional("my_condition")

Calling conditional() pushes a new context onto the FlyteContextManager stack. This is why the call must happen inside a @workflow-decorated function. Outside a workflow context, neither compilation_state nor execution_state is set, and the function raises AssertionError("Branches can only be invoked within a workflow context!").

The fluent chain: .if_(), .elif_(), .else_(), .then(), .fail()

The ConditionalSection exposes a fluent interface through the helper class Condition:

@workflow
def my_workflow(my_input: float) -> float:
    result = (
        conditional("my_condition")
        .if_(my_input > 0.5)
        .then(double(n=my_input))
        .else_()
        .then(square(n=my_input))
    )
    return result

Each of .if_() and .elif_() accepts a comparison or conjunction expression and returns a Case object that captures the branch's expression. Calling .then(output_promise) on the Case registers the branch's output and calls ConditionalSection.end_branch(). If you want the branch to raise an error instead of returning a value, call .fail(err: str) on the Case:

@workflow
def val_wf(my_input: float) -> float:
    return (
        conditional("val")
        .if_(my_input < 0.7)
        .then(double(n=my_input))
        .else_()
        .fail("Only values less than 0.7 allowed")
    )

Expressions: only comparison and conjunction are valid

The expression passed to .if_() or .elif_() must be a ComparisonExpression or a ConjunctionExpression. The Case.__init__ constructor (in condition.py) enforces three guards:

• Already-evaluated booleans are rejected with an AssertionError.
• Raw Promise objects (unary expressions like if_(node_output)) are rejected with an AssertionError.
• Any other type is rejected with an AssertionError.

Comparison expressions use the standard Python operators available on Promise (lines 554–571 of promise.py):

Python expression	Produces
promise == 5	ComparisonExpression(self, ComparisonOps.EQ, 5)
promise > 0.1	ComparisonExpression(self, ComparisonOps.GT, 0.1)
promise.is_true()	ComparisonExpression(self, ComparisonOps.EQ, True)
promise.is_false()	ComparisonExpression(self, ComparisonOps.EQ, False)
promise.is_none()	ComparisonExpression(self, ComparisonOps.EQ, None)

The operators <, >, <=, >=, ==, and != all create ComparisonExpression instances through dunder methods on Promise.

Conjunctions: use & and |, not and / or

To combine two comparison expressions, use the bitwise operators & (AND) and | (OR):

@workflow
def fractions_wf(my_input: float) -> float:
    return (
        conditional("fractions")
        .if_((my_input > 0.1) & (my_input < 1.0))
        .then(double(n=my_input))
        .else_()
        .then(square(n=my_input))
    )

The & and | operators produce ConjunctionExpression objects via ComparisonExpression.__and__ and __or__. Python's built-in and and or keywords trigger __bool__, which both ComparisonExpression and ConjunctionExpression deliberately override to raise ValueError. This is a hard crash, not a warning — always use & and | inside conditional expressions.

Output intersection

When you assign the result of a conditional chain, the variable names available depend on what all branches return. ConditionalSection.compute_output_vars() computes the intersection of output variable names across every branch:

output_vars_set = output_vars_set.intersection(curr_set)

If one branch returns a single int named o0 and another returns a tuple (int, str) named (o0, o1), the intersection is just o0. If any branch returns a VoidPromise or has neither an output_promise nor an err, the entire conditional returns None (wrapped as VoidPromise).

Compilation and execution semantics

Flytekit uses three different ConditionalSection subclasses depending on the runtime mode. The selection happens inside the conditional() factory:

• ctx.compilation_state is set → base ConditionalSection (compilation mode)
• BranchEvalMode.BRANCH_SKIPPED → SkippedConditionalSection
• BranchEvalMode.BRANCH_ACTIVE → LocalExecutedConditionalSection

Compilation mode

When a workflow is being compiled (the normal case during registration), the base ConditionalSection is used. Its end_branch() method:

1. Calls to_branch_node(), which transforms all collected Case objects into an IfElseBlock model via to_ifelse_block(). If there are fewer than two cases, to_ifelse_block() raises AssertionError("At least an if/else is required. Dangling If is not allowed").
2. Creates a BranchNode — a simple wrapper holding the IfElseBlock model.
3. Builds a Node in the compilation state with bindings linking branch outputs to upstream node references.
4. Returns output Promise objects that downstream nodes can reference.

The result is a static DAG node whose flyte_entity is a BranchNode. Every branch's sub-graph is already present in the compiled workflow definition — the runtime engine simply picks one.

Local execution (active branch)

When you run a workflow function locally (in a unit test, for example), flytekit evaluates the expressions and executes the selected branch. LocalExecutedConditionalSection overrides start_branch() to perform short-circuit evaluation:

if self._selected_case is None:
    if c.expr is None or c.expr.eval() or last_case:
        ctx.execution_state.take_branch()
        self._selected_case = added_case

It evaluates c.expr.eval() for each case in order. When it finds a truthy expression (or reaches the else_ case, where c.expr is None), it calls take_branch() on the execution state and records the selected case. Subsequent branches are skipped entirely — their task bodies never run.

When end_branch() fires on the last case, it returns the output from the selected case's then() call. If the selected case has neither an output_promise nor an err, it raises AssertionError("Bad conditional statements, did not resolve in a promise").

Skipped execution (nested conditionals)

When a branch is not selected at runtime, any nested conditionals inside it must not execute their task bodies. SkippedConditionalSection handles this by returning None-filled promises:

promises = [Promise(var=x, val=None) for x in curr]

This ensures that inner tasks are never invoked when their outer branch is skipped.

At least two cases, always terminate with else_

Every conditional chain must have at least one .if_() and one .else_(). The .else_() call always marks itself as last_case=True, which signals end_branch() to finalize the conditional section.

Real examples from the codebase

Condition using input comparison

Assigning the conditional result to a variable and using it in a multi-output return (from workflow.py line 1313):

@workflow
def my_wf_example(a: int) -> typing.Tuple[int, int]:
    x = add_5(a=a)
    z = add_5(a=x)
    d = simple_wf()
    e = conditional("bool").if_(a == 5).then(add_5(a=d)).else_().then(add_5(a=z))
    return x, e

Here a == 5 produces a ComparisonExpression because __eq__ on Promise returns a ComparisonExpression. Both branches call the same task add_5, so their output variable names match and the intersection resolves to a single int.

Nested conditionals with conjunction and .fail()

The docstring at condition.py line 487 demonstrates nesting, conjunction expressions, an .elif_() chain, and .fail() for error branches:

v = (
    conditional("fractions")
    .if_((my_input > 0.1) & (my_input < 1.0))
    .then(
        conditional("inner_fractions")
        .if_(my_input < 0.5)
        .then(double(n=my_input))
        .elif_((my_input > 0.5) & (my_input < 0.7))
        .then(square(n=my_input))
        .else_()
        .fail("Only <0.7 allowed")
    )
    .elif_((my_input > 1.0) & (my_input < 10.0))
    .then(square(n=my_input))
    .else_()
    .then(double(n=my_input))
)

Key points:

• The outer if_ expression uses & to combine (my_input > 0.1) and (my_input < 1.0).
• The then() of the outer if_ receives the result of an entire nested conditional chain. The inner conditional(...) returns Promise objects that the outer .then() captures.
• .else_().fail("...") on the inner chain means inputs between 0.7 and 1.0 cause a terminal error instead of returning a value.
• The outer .else_().then(double(...)) is the catch-all that runs when my_input >= 10.0 or my_input <= 0.1.

Dynamic workflows

The @dynamic decorator is a functools.partial of the @task decorator with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC:

from flytekit import dynamic

dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

Why dynamic exists

A regular @task function runs at execution time but cannot create new workflow nodes — its output is just data. A @workflow function runs at compilation time and can create nodes, but cannot use its inputs as raw Python values (they are Promise objects). A @dynamic function merges both capabilities: its body runs at execution time and can emit new task nodes, effectively producing a subworkflow on the fly.

The docstring at dynamic_workflow_task.py explains:

In short, a task's function is run at execution time only, and a workflow function is run at compilation time only (local execution notwithstanding). A dynamic workflow is modeled on the backend as a task, but at execution time, the function body is run to produce a workflow.

Canonical example: iteration over input

The most common use case is a loop whose iteration count depends on an input — something impossible in a static @workflow because you cannot call Python range() on a Promise:

@dynamic
def my_dynamic_subwf(a: int) -> typing.Tuple[typing.List[str], int]:
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s, 5

This creates a task invocations of t1 at runtime, each with a different argument. The resulting subworkflow contains a nodes.

Expressing dependencies between dynamic tasks

Inside a @dynamic function, you can chain tasks just like in a regular workflow:

@dynamic
def my_dynamic_subwf(a: int, b: int) -> int:
    x = t1(a=a)
    return t2(b=b, x=x)

Here t2 receives the output of t1 as its x argument, creating a data dependency edge in the dynamically generated subworkflow.

Constraints and gotchas

• Keep it small. The module docstring warns: "It's rare to see a manually written workflow that has 5000 nodes ... but you can easily get there with a loop. Please keep dynamic workflows to under fifty tasks." Large dynamic subworkflows strain the Flyte propeller compiler.
• Node dependency hints. Because the Flyte engine cannot know before execution which tasks a @dynamic function will call, you may need to declare them upfront via the node_dependency_hints parameter on PythonFunctionTask. These hints help the platform resolve launch plan registrations for referenced tasks, workflows, or launch plans.
• No manual create_node() inside skipped branches. Flytekit explicitly disallows manual node creation when branch_eval_mode == BRANCH_SKIPPED, preventing orphaned nodes in branches that should never run.

Imperative workflow construction

If you are building workflows imperatively — without the @workflow decorator — WorkflowBase.create_conditional() at workflow.py line 589 provides programmatic access:

def create_conditional(self, name: str) -> ConditionalSection:
    ctx = FlyteContext.current_context()
    if ctx.compilation_state is not None:
        raise RuntimeError("Can't already be compiling")
    FlyteContextManager.with_context(ctx.with_compilation_state(self.compilation_state))
    return conditional(name=name)

It sets up a fresh compilation state and delegates to the same conditional() factory, giving you the same fluent API.

Summary of rules

Rule	What happens if violated
Use & and `	instead ofandandor`
Expressions must be ComparisonExpression or ConjunctionExpression	AssertionError in Case.__init__
No unary if_(node_output) — always compare	AssertionError in Case.__init__
Every chain needs at least if_ plus else_	AssertionError in to_ifelse_block()
Must be called inside @workflow context	AssertionError in conditional()
Keep dynamic workflows under ~50 nodes	Warning in module docstring