Conditional and dynamic workflows

Choose between a conditional and a dynamic workflow

Use conditional(...) when the branches are part of the workflow graph you author: Flytekit compiles the chain into an IfElseBlock. Use @dynamic when execution-time Python values determine which tasks to create or how many tasks to create. These are different execution models. In particular, eager workflows do not support Flytekit conditionals; their source explicitly recommends a plain Python if instead.

A typed conditional workflow

Import the public factory from flytekit.core.condition, and return the result of the branch chain rather than writing a Python if around Flyte promises:

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())  # type: ignore


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

a == True is a Flyte comparison expression during workflow construction, not an evaluated branch in the compiled workflow. Both branches return bool, so the conditional can provide a typed result. The branch syntax is fluent:

• conditional("name") creates a conditional section.
• .if_(expression) starts the first Case.
• .then(promise) records that case's output and closes it.
• .elif_(expression) starts another comparison or conjunction case.
• .else_() starts the terminal case and marks it as the last case.
• .fail("message") records a terminal failure instead of an output.

There is no implicit terminal branch. A chain that ends after if_() is rejected during conversion to an IfElseBlock; terminate the chain with .else_() (or with a supported elif_() followed by .else_()).

Expressions and branch outputs

Keep conditions as Flyte expression objects. Case accepts ComparisonExpression and ConjunctionExpression, including comparisons such as <, <=, >, >=, ==, and !=, and conjunctions formed with & or |. Python and, or, is, and not evaluate in Python and do not form the expression objects required by the API. An already evaluated bool, a unary promise such as if_(x), an arbitrary expression type, or None is rejected.

For example, the source uses a primitive comparison against a workflow input and assigns task outputs from both sides:

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 remains a comparison expression during workflow construction. In a typed workflow, make the output shape of each branch compatible. ConditionalSection.compute_output_vars() intersects the promise variable names contributed by all cases. If a case is void, has neither an output nor an error, or leaves no common output, the conditional is represented as void rather than exposing an incompatible result.

Use .fail(...) when a branch is intentionally unsuccessful:

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

At compilation, the final failure becomes the serialized condition model's Error. During local execution, selecting that failed case raises ValueError; a selected case with neither a promise nor an error raises an assertion.

What compilation produces

conditional(name) requires an active Flyte workflow context. In compilation mode it creates a ConditionalSection. Its constructor creates the case list and pushes a child context with enter_conditional_section(). Each Case.then() or Case.fail() delegates to end_branch(); when the final case completes, ConditionalSection.end_branch() pops that context and performs the graph conversion.

The compilation path is effectively:

conditional -> ConditionalSection -> Case chain
                                      -> to_ifelse_block()
                                      -> BranchNode / IfElseBlock
                                      -> graph Node

The resulting BranchNode stores the conditional name and serialized _core_wf.IfElseBlock. end_branch() wraps it in a Node, gives it NodeMetadata, collects referenced promise bindings, and adds upstream nodes to the workflow compilation state. transform_to_comp_expr() and transform_to_conj_expr() turn the Flyte expression tree into model expressions. A promise operand is bound using a generated node_id.var name; merge_promises() deduplicates references and avoids collisions when different nodes expose similarly named outputs.

The promise returned to later workflow code is therefore a graph output, not the concrete value of the branch during compilation. Its variables are limited to the common output variables calculated by compute_output_vars(). This is why matching branch output shapes matters even when only one branch executes at runtime.

Local execution and nested branches

Local execution uses LocalExecutedConditionalSection, which evaluates each case expression while visiting the fluent chain. It selects the first case whose expression evaluates true, or the final case as the fallback, and calls ExecutionState.take_branch(). Each branch completion calls branch_complete(). Once the final case is reached, the selected case's concrete values are wrapped in promises and returned.

Nested conditionals have an additional local-execution behavior. If an outer branch is inactive, Flytekit creates SkippedConditionalSection for a nested conditional. That section preserves the chaining API but returns placeholder promises (or a VoidPromise when there are no outputs), so tasks in the inactive nested branch are not executed. The nested example above consequently remains usable as the outer .then(...) value while still obeying the common-output rule.

Conditional construction is valid only inside a workflow context. Calling conditional() outside compilation or workflow execution raises AssertionError("Branches can only be invoked within a workflow context!"). The context is managed by the Flyte context stack rather than a Python with block; final branch completion pops the conditional context. Calls that manually create nodes in skipped branch logic are also rejected by node creation with RuntimeError.

Dynamic workflows are runtime graph generation

Choose @dynamic when a runtime-native input controls graph generation. The dynamic decorator in dynamic_workflow_task.py is defined as a partial 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

A dynamic function runs at execution time and produces a workflow that Flyte runs as a subworkflow. Unlike an ordinary workflow function, it can use its inputs as native Python values, so range(a) is valid in this example. It can also construct dependencies among generated tasks:

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

The dynamic workflow source cautions that the generated workflow should stay under roughly fifty tasks; for large-scale identical fan-out it recommends a map task. Dynamic tasks cannot use reference tasks in the inspected implementation. If a dynamic function uses a launch plan dynamically, supply it through node_dependency_hints; those hints are intended for dynamic tasks and the launch plan must already be registered.

Choice	When the function/body is used	Resulting behavior
conditional(...)	Workflow compilation, with Flyte promise expressions	A static graph Node containing an IfElseBlock; local execution selects a case and skips inactive nested work
@dynamic	Execution time, with native runtime inputs	A task-like backend entity that generates and returns a subworkflow
Eager workflow	Python async execution	Conditionals are unsupported; use a plain Python if

WorkflowBase.create_conditional(name) is the workflow integration point: it temporarily installs workflow compilation state and delegates to conditional(name), and refuses to run while the workflow is already compiling.