Conditional and dynamic workflows

Conditional branches and dynamic workflows

A plain Python if inside a @workflow function does not create a runtime branch. Flytekit evaluates the workflow body once at compile time to build a static DAG, so a native if would execute immediately against placeholder Promise objects and either raise or hard-code the branch choice. Flytekit solves this with conditional(), which compiles into a BranchNode wrapping an IfElseBlock that the Flyte engine evaluates later. When the shape of the DAG itself depends on runtime values—iterating over a list whose length is unknown at registration time, for example—@dynamic lets you generate the subgraph at execution time.

Using conditional() inside a workflow

Call conditional() only inside a @workflow or @dynamic function. In flytekit/core/condition.py the factory inspects the current context and returns ConditionalSection when compiling, LocalExecutedConditionalSection when running locally, or SkippedConditionalSection for nested branches that are already skipped. If you call it outside a workflow context it raises an AssertionError:

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

The DSL is fluent: conditional(name).if_(expr).then(task_call)… optionally more .elif_(expr).then(task_call)… and finally .else_().then(task_call). You must provide the .else_() clause. If you omit it, flytekit/core/workflow.py rejects the workflow during output binding with:

if isinstance(workflow_outputs[i], ConditionalSection):
    raise AssertionError(
        "A Conditional block (if-else) should always end with an `else_()` clause"
    )

Here is a typed workflow from flytekit/core/workflow.py that branches on an integer input:

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

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

@workflow
def simple_wf() -> int:
    return add_5(a=1)

@workflow
def my_wf_example(a: int) -> 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

How conditionals compile

When compilation_state is active, conditional() returns a ConditionalSection. Its __init__ pushes a new context onto the FlyteContextManager so that task calls inside the branch are captured separately:

ctx = FlyteContextManager.current_context()
FlyteContextManager.push_context(ctx.enter_conditional_section().build())

Each .if_(), .elif_(), and .else_() creates a Case and invokes start_branch(). Case.then() stores the output promise and calls end_branch(). On the last case, ConditionalSection.end_branch() builds a BranchNode via to_branch_node and adds the resulting node to the compilation state:

node, promises = to_branch_node(self._name, self)
...
n = Node(
    id=f"{ctx.compilation_state.prefix}n{len(ctx.compilation_state.nodes)}",
    metadata=_core_wf.NodeMetadata(self._name, timeout=datetime.timedelta(), retries=RetryStrategy(0)),
    bindings=sorted(bindings, key=lambda b: b.var),
    upstream_nodes=list(upstream_nodes),
    flyte_entity=node,
)
FlyteContextManager.current_context().compilation_state.add_node(n)

The underlying protobuf model is an IfElseBlock produced by to_ifelse_block in flytekit/core/condition.py. The first case becomes case, intermediate cases become other, and the final else_ becomes else_node.

Branches must agree on outputs. ConditionalSection.compute_output_vars() computes the intersection of output variable names across every case:

output_vars_set = output_vars_set.intersection(curr_set)
new_output_var = []
for v in output_vars:
    if v in output_vars_set:
        new_output_var.append(v)
output_vars = new_output_var

If any branch returns a VoidPromise or None, the entire conditional defaults to a void output.

Local execution semantics

When you run a workflow locally—my_wf_example(a=1), for instance—conditional() returns a LocalExecutedConditionalSection. Instead of building a BranchNode, it evaluates the Python expression immediately.

LocalExecutedConditionalSection.start_branch() calls expr.eval() to decide whether the branch is taken:

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

take_branch() sets branch_eval_mode to BRANCH_ACTIVE on the ExecutionState. Once the body finishes, end_branch() calls branch_complete(), switching the mode to BRANCH_SKIPPED:

ctx.execution_state.branch_complete()

Later branches are skipped, preventing their tasks from executing.

For nested conditionals where the outer branch evaluated to false, conditional() returns SkippedConditionalSection. It never evaluates inner expressions and returns placeholder promises with None values:

if curr is None:
    return VoidPromise(self.name)
promises = [Promise(var=x, val=None) for x in curr]
return create_task_output(promises)

Writing valid branch expressions

Promise objects override comparison operators to return ComparisonExpression instances rather than plain booleans. In flytekit/core/promise.py:

def __eq__(self, other) -> ComparisonExpression:
    return ComparisonExpression(self, ComparisonOps.EQ, other)

def __gt__(self, other) -> ComparisonExpression:
    return ComparisonExpression(self, ComparisonOps.GT, other)

Chain comparisons with the bitwise operators & and | to build ConjunctionExpression objects:

(my_input > 0.1) & (my_input < 1.0)

Both ComparisonExpression and ConjunctionExpression define __and__ and __or__ for chaining. Their __bool__ methods raise ValueError to stop Python’s and and or from evaluating the expression early:

def __bool__(self):
    raise ValueError(
        "Cannot perform truth value testing,"
        " This is a limitation in python. For Logical `and\or` use `&\|` (bitwise) instead."
    )

Case.__init__ enforces three more rules.

No bare booleans. Using and/or evaluates to bool and triggers:

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

No unary promises. Passing a plain Promise like if_(my_input) raises:

raise AssertionError(
    "Flytekit does not support unary expressions of the form `if_(x) - where x is an"
    " input value or output of a previous node."
)

Primitives only. ComparisonExpression rejects non-primitive Promise values:

raise ValueError("Only primitive values can be used in comparison")

Nested conditionals

Conditionals can be nested. The docstring for conditional() in flytekit/core/condition.py shows an outer branch that contains another conditional:

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

If the outer if_ is false, the inner conditional is replaced by a SkippedConditionalSection, so tasks inside it are never invoked during local execution.

Limitations inside branch logic

Manual node creation with create_node() is disallowed inside conditional branches during local execution. flytekit/core/node_creation.py raises a RuntimeError when branch_eval_mode is BRANCH_SKIPPED:

if ctx.execution_state.branch_eval_mode == BranchEvalMode.BRANCH_SKIPPED:
    raise RuntimeError(
        "Being more restrictive for now and disallowing manual node creation in branch logic"
    )

Generating workflows at runtime with @dynamic

A @workflow body is compiled once at registration time; you cannot use Python loops or range() that depend on an input value. @dynamic bridges the gap. It is defined in flytekit/core/dynamic_workflow_task.py as:

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

The decorated function runs at execution time to produce a workflow definition, which the Flyte engine then executes as a subworkflow.

from typing import List
from flytekit import dynamic, task

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

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

Notice that range(a) is valid because the function body executes with the real integer value of a before the resulting graph is submitted.

In flytekit/core/python_function_task.py, compile_into_workflow handles remote execution. It creates a fresh compilation state with prefix "d", runs the user function through a PythonFunctionWorkflow, serializes the result with get_serializable, and returns a DynamicJobSpec:

es = ctx.new_execution_state().with_params(mode=ExecutionState.Mode.DYNAMIC_TASK_EXECUTION)
updated_ctx = updated_ctx.with_execution_state(es)
with FlyteContextManager.with_context(updated_ctx):
    self._create_and_cache_dynamic_workflow()
    cast(PythonFunctionWorkflow, self._wf).compile(**kwargs)
    ...
    dj_spec = _dynamic_job.DynamicJobSpec(
        min_successes=len(workflow_spec.template.nodes),
        tasks=tts,
        nodes=workflow_spec.template.nodes,
        outputs=workflow_spec.template.outputs,
        subworkflows=workflow_spec.sub_workflows,
    )

For local runs, dynamic_execute detects is_local_execution() and runs the function directly in LOCAL_DYNAMIC_TASK_EXECUTION mode, translating the returned Python values back into a LiteralMap.

How compilation and execution semantics differ

Feature	Conditional (conditional())	Dynamic (@dynamic)
When the structure is fixed	At compile time; branches are static IfElseBlock nodes	At execution time; Python loops and ifs generate the graph
Input-dependent control flow	Runtime choice via IfElseBlock	Native Python control flow (range, if, for)
Output shape contract	Intersection of output variable names across all cases	Determined by the Python function at runtime
Backend representation	BranchNode / IfElseBlock	DynamicJobSpec containing generated nodes and tasks

You can also use conditional() inside a @dynamic workflow. During remote execution the dynamic task compiles its inner workflow, and the conditional becomes a static BranchNode within that generated subgraph. During local execution, the dynamic function runs as normal Python and the conditional short-circuits branches exactly as it does in a static workflow.