Skip to main content

Conditional and dynamic workflows

Choose a conditional branch or a dynamic workflow​

Use conditional(...) when the workflow graph is known while flytekit compiles the @workflow function, but a Flyte input or upstream promise determines which already-described branch runs. Do not replace it with a Python if: workflow inputs and task outputs are represented by Flyte expressions, and the conditional expression must remain available to compilation.

Use @dynamic when the workflow graph itself must be generated from runtime values. A dynamic function runs at execution time and can use its inputs as native Python values—for example, range(a)—to decide how many task calls to create. The two features therefore operate at different levels:

  • conditional describes an IfElseBlock while a static workflow is compiled (and evaluates the same fluent structure during local execution).
  • dynamic is modeled as a task initially; at execution time its function body generates a workflow that Flyte runs as a subworkflow.

Express a static branch with the fluent API​

Write a branch as if_(...).then(...), optionally continue with elif_(...).then(...), and finish with else_().then(...) or else_().fail(...). The branches return a value through the conditional expression, so give the enclosing workflow a return annotation that matches that value:

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

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

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

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

assert wf() is True
assert wf(a=False) is False

This example is taken from the workflow implementation's local-execution coverage. The a == True expression is a Flyte comparison expression, not a Python if; local execution evaluates it and the two assertions observe the selected task's value.

A comparison against a literal works with integer inputs and task promises as well. my_wf_example returns the conditional result alongside another task output, so its annotation is typing.Tuple[int, int]:

import typing

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

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

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

@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

Condition is the fluent controller returned by conditional(...). Its if_() and elif_() methods create Case objects, while else_() creates the final case. A Case.then(p) records the branch output and closes that branch; Case.fail(err) records an error and also closes it. Consequently, the chain must end with a final else_() branch. A dangling ConditionalSection is rejected by workflow output binding, and conversion to an IfElseBlock requires at least two cases.

Conditions accept comparison operators (<, <=, >, >=, ==, and !=) and conjunctions made with & or |. Combine the expressions rather than using Python boolean operators. For example, the conditional factory's documented nested form uses a conjunction and an error case:

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

Here the inner conditional is itself the output of the outer then(...), which is how nested conditionals are expressed. The my_input, double, and square names are the task/input names used by the condition.py docstring example; their types must be compatible with the comparisons and task signatures in the workflow where you use this pattern.

What static compilation produces​

When conditional(name) sees a context with compilation state, it constructs a ConditionalSection. Its constructor creates the Condition chain and pushes a context marked as being in a conditional section. Intermediate then(...) calls return the chain so that elif_() or else_() can be appended.

The final then(...) or fail(...) invokes ConditionalSection.end_branch(). For the last case, that method pops the conditional context, calls to_branch_node(self._name, self), and receives a BranchNode plus the promises referenced by the branches. It then creates a Node whose flyte_entity is the branch node, creates bindings for promises that are not ready, adds referenced nodes as upstream nodes, registers the node in the compilation state, and returns the conditional outputs. BranchNode associates the conditional name with the backend _core_wf.IfElseBlock; its public name property exposes that name.

The compilation flow can be summarized as:

Flyte comparisons/conjunctions
        -> model BooleanExpression / IfBlock
Condition + Case chain
        -> to_ifelse_block(...)
BranchNode(name, IfElseBlock)
        -> Node(bindings, upstream_nodes, flyte_entity=BranchNode)
        -> workflow compilation state

Expression transformation converts a Promise into a qualified node_id.var operand and converts literals into primitive/scalar operands. Comparison operators become model comparison expressions, while & and | become model conjunction expressions. The resulting backend block represents the branch graph; flytekit does not evaluate the workflow input as a Python if during static compilation.

Branch outputs must have a compatible common interface. ConditionalSection.compute_output_vars() intersects the output variable names found in all cases. _compute_outputs() exposes promises for that common set. If a case has no output, has an error, or produces a VoidPromise, the conditional is treated as having no usable value and returns a VoidPromise. A final .fail("...") is serialized as the conditional error path rather than an output value.

These rules also apply to output shapes such as tuples or named-task outputs: each branch must provide the common output variables that the enclosing workflow consumes. A branch that returns a different set cannot provide an independent output shape merely because that branch is not selected at runtime.

Local execution and skipped branches​

The same fluent expression can be called while a workflow is locally executed. The conditional factory selects the implementation from the current FlyteContext:

  • compilation state selects ConditionalSection;
  • local execution selects LocalExecutedConditionalSection;
  • local execution whose branch evaluation mode is BRANCH_SKIPPED selects SkippedConditionalSection;
  • no applicable workflow context raises AssertionError("Branches can only be invoked within a workflow context!").

LocalExecutedConditionalSection.start_branch() evaluates each case expression with expr.eval() until a case is selected. An unconditional else_() or the final case acts as the fallback. Once selected, it calls ExecutionState.take_branch(). After each branch, end_branch() calls ExecutionState.branch_complete(). Task/entity calls made while the branch state is skipped are intercepted by promise handling and become None-valued promises (or a VoidPromise) rather than executing the task.

On the final branch, local execution returns the selected branch's actual values, but maps them onto the least-common output variables calculated across all cases. If the selected case has neither an output promise nor an error, flytekit raises an assertion; if it recorded an error with fail(...), local execution raises that error as a ValueError after the final branch.

Nested conditionals in an inactive local branch use SkippedConditionalSection. It continues visiting the fluent syntax so the chain can be completed, but does not execute the nested branch tasks. Its final result contains None-valued promises for common outputs, or a VoidPromise when there are no usable outputs. Ordinary task calls are therefore safe to represent in skipped syntax, but manual node creation is explicitly rejected in skipped branch logic with a RuntimeError.

Generate a runtime workflow with @dynamic​

Choose @dynamic instead of conditional when the runtime input determines the number or arrangement of nodes. In flytekit/core/dynamic_workflow_task.py, dynamic is defined as a partial application of task.task with PythonFunctionTask.ExecutionBehavior.DYNAMIC:

@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

The range(a) call is valid in this dynamic function because its body runs at execution time and a is a native Python value there. A dynamic function can also express dependencies between the generated task calls:

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

This differs from a static @workflow: flytekit's dynamic-workflow documentation states that a workflow function runs at compilation time (apart from local execution), while a dynamic function runs at execution time to produce a workflow. The backend initially models the dynamic workflow as a task, then runs the generated workflow as a subworkflow. In contrast, conditional compiles one branch node containing the cases and returns promises from its common outputs.

Keep generated dynamic workflows under roughly fifty tasks. dynamic_workflow_task.py notes that a loop can easily generate thousands of nodes, which must still be processed like any other workflow, and recommends map tasks for large-scale identical runs.

Troubleshooting constraints​

  • Python and, or, is, or not: these can produce an evaluated boolean or an unsupported expression. Use Flyte comparisons and &/| conjunctions.
  • if_(promise): raw unary Promise conditions are rejected. Compare the promise with a supported literal or form a supported conjunction.
  • Missing else_(): finish every conditional chain. The final case is required for the backend IfElseBlock and for workflow output binding.
  • Mismatched branch outputs: return compatible common outputs from every successful branch; flytekit intersects output variable names rather than choosing a branch-specific shape.
  • Conditional outside a workflow: call conditional(...) only while flytekit has compilation or local execution state; otherwise the factory raises the documented assertion.
  • Manual node creation in a skipped local branch: ordinary task/entity calls are converted to skipped promises, but manual node creation is rejected by flytekit's node-creation path.
  • Interrupted fluent chains: construction pushes a conditional context and completion pops it. If compilation or evaluation fails mid-chain, flytekit's surrounding context management includes cleanup for leaked conditional contexts.