Skip to main content

Conditional and dynamic workflows

Conditional branches and dynamic workflows​

Flytekit gives you two different ways to shape the control flow of a workflow — and they are easy to conflate because both look like plain Python inside the body of a function. The first, the conditional block, is resolved at compile time into a static branch node. The second, @dynamic, produces a task whose function body runs at execution time to generate a sub-workflow. Choosing the right one depends on whether you know your branch conditions before the workflow runs, and whether you need to loop over native Python values.

  • Use conditional(...) when the branches are known up front and you want the engine to pick one branch statically.
  • Use @dynamic when you need to iterate over actual values (a real range, a real list) and let the loop body "unroll" into nodes at runtime.

Both live in flytekit.core — condition.py for the conditional machinery and dynamic_workflow_task.py for the dynamic decorator.

The conditional fluent API​

The entry point is the factory function conditional(name) in flytekit/core/condition.py. Calling it returns a "section" that lets you build a ternary-style if/else chain: it behaves like a function, so the branch you pick is the value the whole expression evaluates to. The canonical form is

from flytekit import conditional, task, workflow

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

@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

(This is adapted from the real workflow at workflow.py:1314. Note that the branch bodies — add_5(a=d) and add_5(a=z) — are task calls, not plain values, so the conditional routes between nodes rather than between constants.)

The chain is built from three pieces:

  • Condition (condition.py:326) — returned by conditional(...), it exposes .if_(expr), .elif_(expr), and .else_(). Each call registers a Case in the enclosing section.
  • Case (condition.py:242) — one clause. .if_/.elif_ wrap a comparison or conjunction expression; else_ carries no expression. .then(promise) supplies the branch's output promise, and .fail(err) records an error message instead.
  • ConditionalSection (condition.py:35) — the container that accumulates the Case objects and eventually emits the compiled branch node.

When a boolean condition comes from another task, use the is_true() / is_false() helpers on the Promise rather than == True:

from typing import Any
from flytekit import conditional, task, workflow

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

@task
def f() -> bool:
    return False

@workflow
def wf(a: Any = True) -> bool:
    return conditional("bool").if_(a.is_true()).then(t()).else_().then(f())

This comes from the test_default_values workflow in workflow.py:1056 (there, the parameter is annotated a: bool = True, but inside the workflow body a is really a Promise). is_true() is defined on Promise (promise.py:548) and returns ComparisonExpression(self, ComparisonOps.EQ, True).

The expressions you pass to .if_/.elif_ must be one of the two legal types:

  • ComparisonExpression — produced by the rich comparison operators on Promise (==, !=, >, >=, <, <=), each of which returns a comparison expression rather than a bool. Promise.__eq__, __gt__, etc. are all defined in promise.py:554-570.
  • ConjunctionExpression — produced by combining two expressions with the bitwise & and | operators (ComparisonExpression.__and__ / __or__ at promise.py:352-356).

You can't use Python's and / or here; Promise.__and__ and __or__ deliberately raise ValueError (promise.py:578-582), and __bool__ raises ValueError too (promise.py:572-576), which is precisely why ConditionalSection's docstring shows the & form:

v = (
    conditional("fractions")
    .if_((my_input > 0.1) & (my_input < 1.0))
    .then(...)
    .elif_((my_input > 1.0) & (my_input < 10.0))
    .then(square(n=my_input))
    .else_()
    .then(double(n=my_input))
)

What the two execution modes do differently​

The same conditional(name) call resolves to a different ConditionalSection subclass depending on which flytekit context it runs under. The dispatch lives in the factory itself (condition.py:509-522):

if ctx.compilation_state:            -> ConditionalSection            (compile into static branch)
elif ctx.execution_state (local):    -> LocalExecutedConditionalSection (evaluate the branch now)
     ... branch_eval_mode == SKIPPED -> SkippedConditionalSection     (short-circuit a nested branch)
raise AssertionError("Branches can only be invoked within a workflow context!")

That final AssertionError is why calling conditional(...) outside of a @workflow body fails — the machinery needs the workflow context to exist.

Compilation mode. When you define a @workflow and flytekit scans it (or serializes it), there is a compilation_state, so conditional returns ConditionalSection. Its constructor pushes a child context via FlyteContextManager.push_context(ctx.enter_conditional_section().build()), and each .then() / .fail() triggers end_branch() (condition.py:86). On the final case, end_branch() pops the context and collapses the whole chain into a single Node whose flyte_entity is a BranchNode:

node, promises = to_branch_node(self._name, self)
n = Node(id=..., metadata=..., bindings=..., upstream_nodes=..., flyte_entity=node)

to_branch_node (condition.py:473) delegates to to_ifelse_block (condition.py:445), which builds a serializable IfElseBlock protobuf — one IfBlock per case plus an else_node (or an error when the last case was a .fail()), stored inside a BranchNode (condition.py:25). In other words, at compile time flytekit does not evaluate the condition; it emits an IfElseBlock describing all branches, and the engine picks one when the workflow actually runs.

Local execution mode. When the workflow is invoked locally (e.g. wf(a=5) in a test), there is an execution_state but no compilation_state, so conditional returns LocalExecutedConditionalSection (condition.py:163). This subclass actually evaluates the expression. In start_branch() it checks c.expr.eval() — ComparisonExpression.eval() (promise.py:335) and ConjunctionExpression.eval() (promise.py:403) compute a real bool from the concrete values. If the clause is truthy and no branch has been chosen yet, it calls ctx.execution_state.take_branch() — which sets BranchEvalMode.BRANCH_ACTIVE (context_manager.py) — and records the case as _selected_case.

Crucially, end_branch() calls ctx.execution_state.branch_complete() after every branch, flipping the mode to BRANCH_SKIPPED. So once the winning branch is chosen, any task inside a later, unselected branch is never actually invoked. When the final case is reached, end_branch() returns the Promise outputs captured from _selected_case rather than a compiled node. That's how assert my_wf_example(a=1) == (6, 16) (workflow.py:1341) resolves to real Python values.

Skipped mode. For a conditional nested inside an already-skipped branch, conditional returns SkippedConditionalSection (condition.py:219). Its end_branch() never evaluates any expression — it just pops the context and returns a VoidPromise (or promises initialized to None) so the outer code can consume a value without running any branch-body tasks. This keeps nested conditionals consistent when an outer branch was decided false.

Each branch must contribute the same outputs​

Because the engine picks the branch at runtime, the conditional's type must be decidable at compile time. ConditionalSection.compute_output_vars() (condition.py:141) intersects the names of the output variables across every case, keeping only the variables present in all of them:

output_vars_set = output_vars_set.intersection(curr_set)

If any case contributes no output (or returns a VoidPromise), the intersection is empty and the whole conditional returns a VoidPromise — the branches are treated as side-effect-only. This is why the branch bodies above all return a single int: the intersection of {int-output} across branches is that same variable, and the workflow can use e as a typed value. Keep every then(...) producing the same shaped output (single value, or identical named-tuple fields) and the conditional stays typable.

.fail() versus .then() on the last case​

The final clause can terminate in either then(...) or fail(...). Both funnel through Case.then / Case.fail into end_branch(), but they produce different IfElseBlocks at compile time. In to_ifelse_block (condition.py:460-466):

if last_case.output_promise is not None:
    node = last_case.output_node
else:
    err = Error(failed_node_id=node_id, message=last_case.err if last_case.err else "Condition failed")

A then becomes the else_node; a fail becomes an error on the block, so reaching that last branch raises the error rather than returning a value. The docstring example shows this — the nested conditional("inner_fractions") ends with .else_().fail("Only <0.7 allowed") (condition.py:500), while the outer one ends with .else_().then(double(n=my_input)).

Compile-time constraints worth knowing​

The Case constructor (condition.py:242) validates its expression up front and raises AssertionError for the three forms flytekit will not compile:

  • a bool — "Logical (and/or/is/not) operations are not supported..."
  • a bare Promise — unary if_(x) where x is a workflow value isn't supported
  • anything that isn't a ComparisonExpression or ConjunctionExpression

A dangling if_ is also rejected. to_ifelse_block raises AssertionError("At least an if/else is required. Dangling If is not allowed") if fewer than two cases exist (condition.py:448), and the workflow compiler raises "A Conditional block (if-else) should always end with an \else_()` clause"(workflow.py:899) if a conditional section never resolves to a final output. In practice: always end the chain with.else_()`, and if you return a conditional from a workflow that declares outputs, make sure the section resolves to a promise.

One more mechanic worth knowing: to build the bindings feeding the branch conditions, flytekit can't read your Python LHS variable names, so create_branch_node_promise_var (condition.py:364) names each operand f"{node_id}.{var}" to keep referenced outputs unique across the workflow.

Contrast: @dynamic for runtime-generated workflows​

Where a conditional decides between a fixed set of statically compiled branches, a dynamic workflow defers the structure of the workflow to execution time. dynamic is defined in dynamic_workflow_task.py:21:

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

It's literally the @task decorator with the DYNAMIC execution behavior, so a dynamic entity is modeled on the backend as a task. But at execution time its function body runs and emits a workflow, which the engine then runs as a sub-workflow. The docstring (dynamic_workflow_task.py:27-30) says it plainly: "a task's function is run at execution time only, and a workflow function is run at compilation time only... It is almost as if the decorator changed from @task to @workflow except workflows cannot make use of their inputs like native Python values whereas dynamic workflows can."

That last point is the key distinction and it shows up in the docstring example:

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

Inside a normal @workflow, calling range(a) on a workflow input fails because the input is a Promise, not an int. Inside @dynamic, a is a real native value at runtime, so range(a) is legal and the loop unrolls into as many t1 nodes as the runtime value demands. You can also express inter-task dependencies directly:

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

Neither of these is possible in a plain workflow body. The cost is that a dynamic workflow is processed like any other workflow after it's produced, so the docstring urges keeping dynamic workflows small — "dynamic workflows to under fifty tasks" (dynamic_workflow_task.py:12) — since a loop can trivially balloon into thousands of nodes, which the compiler then has to handle.

The rule of thumb, then: if your branching depends on values that only materialize when the workflow runs and that force iteration or structure you can't statically enumerate, reach for @dynamic. If you have a known, finite set of branches, a conditional block keeps them statically compiled and the engine routes between them cheaply.