Conditional and dynamic workflows

Choosing between conditional branches and dynamic workflows

Use a conditional branch when the set of tasks and dependencies is known while an @workflow is being compiled, but the value of a typed comparison should select which result is used. Use a dynamic workflow when the workflow graph itself must be generated at execution time—for example, when a native input controls a Python range loop.

These are separate mechanisms in flytekit: conditional() creates a branch node in condition.py, while dynamic is a task decorator configured in dynamic_workflow_task.py.

Conditional branches

Write a conditional as a fluent chain inside a workflow. Each branch must finish with .then(...) or .fail(...), and the chain must include an if_ and a final else_:

@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

The same conditional result can be composed with other workflow outputs. This example from workflow.py compares an integer input and returns the conditional result alongside an ordinary task result:

@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

conditional(name) is only valid inside a workflow context. Its factory inspects the current FlyteContext: compilation returns ConditionalSection; local execution returns LocalExecutedConditionalSection; and an inactive nested local branch returns SkippedConditionalSection. Without compilation or execution state it raises AssertionError with the message Branches can only be invoked within a workflow context!.

How compilation builds the branch node

WorkflowBase.create_conditional() installs the workflow compilation state in the current context and delegates to conditional(name=...). Constructing ConditionalSection then pushes a context marked as a conditional section. The fluent objects have these responsibilities:

• Condition controls the chain. _if() and elif_() create a Case with an expression; else_() creates the final expression-less case and marks it as the last case.
• Case.then(p) records the branch output and calls end_branch(). .fail(err) records an error and also ends the branch.
• ConditionalSection.end_branch() returns the chain condition for intermediate branches. On the final branch it pops the conditional context, calls to_branch_node(), creates a normal core Node, and adds that node to the containing compilation state's node list.
• BranchNode associates the conditional name with the generated core IfElseBlock. The branch wrapper is stored as the flyte_entity of the normal Node created by ConditionalSection.

Condition expressions are transformed from promise expressions into model condition objects. Promise operands are represented using node_id.var; create_branch_node_promise_var() uses this qualified form so identically named outputs from different nodes do not collide. The resulting node also receives bindings for condition-input promises and upstream nodes collected from their references.

Compilation requires a real if/else structure. to_ifelse_block() rejects an empty case list and rejects fewer than two cases with Dangling If is not allowed. A terminal .fail(message) becomes a core Error in the serialized IfElseBlock, using the conditional node ID.

Branch output compatibility

A conditional exposes only output variables common to all cases. ConditionalSection.compute_output_vars() intersects the variable names from each case. If a case has no promise or error, or returns a VoidPromise, the conditional is treated as void. Consequently, give corresponding branches compatible typed outputs when the conditional is used as a workflow value; use .fail(...) when a branch should terminate with an error instead of producing a value.

The implementation also accepts tuple-like and named-tuple-like outputs. Case.then() records a Promise, tuple, VoidPromise, or an object with _fields, and attempts to identify the originating node. Ready/native values do not necessarily have a node pointer, so the node association is only available when the returned promises reference a node.

Local execution and skipped branches

Local execution does not execute both alternatives and then choose between their values. LocalExecutedConditionalSection.start_branch() evaluates each ComparisonExpression or ConjunctionExpression with c.expr.eval() until it selects a case. It calls ExecutionState.take_branch() for the selected case. Every branch completion calls ExecutionState.branch_complete(); on the final branch, the selected case's local values are returned. A selected .fail(err) raises ValueError, while an unresolved selected case raises an assertion.

When a nested conditional is reached from an inactive branch, the factory creates SkippedConditionalSection. Its final end_branch() returns a void promise or promises containing None for the common output variables. The execution and promise/node-creation code uses the skipped branch mode so tasks in that inactive branch return placeholders rather than executing user code. This is why the boolean example's two local assertions can select different task results without running both branch tasks.

Condition syntax and type checks

Use only comparison operators <, <=, >, >=, ==, and !=, or conjunction operators & and |. Case rejects an already evaluated Python bool, a unary Promise condition such as if_(x), and other expression types. Python and, or, is, and not can evaluate before flytekit receives the expression and are therefore not valid conditional inputs. Compare a promise with a supported value, as in a == 5 above.

Conjunctions, nested conditionals, elif_, and terminal failure can be combined. The following is the module-level example in condition.py; my_input, double, and square must be defined as compatible inputs/tasks in the surrounding workflow:

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

Dynamic workflows are a different execution model

Choose @dynamic when Python execution must generate the task graph. In dynamic_workflow_task.py, dynamic is defined as:

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

A dynamic function runs at execution time and is modeled by the backend as a task; its body generates a workflow that Flyte runs as a subworkflow. Unlike an ordinary workflow body, a dynamic function can use inputs as native Python values:

@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

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

The first function constructs a variable number of t1 calls from range(a); the second demonstrates dependencies between dynamically created tasks. These are not ConditionalSection subclasses and do not serialize an IfElseBlock through condition.py.

Requirement	conditional()	@dynamic
Graph construction	Workflow compilation records both branch alternatives in an IfElseBlock and a branch Node.	The decorated function runs at execution time and generates a workflow.
Decision/input behavior	Uses typed promise comparisons or conjunctions such as a == 5.	Can use inputs as native Python values, including range(a).
Runtime behavior	Local execution selects a case and skips inactive tasks; compiled execution evaluates the serialized branch block.	Backend models the function as a task, then runs the generated workflow as a subworkflow.
Typical choice	A fixed graph with alternative typed results or a controlled failure branch.	A graph whose number or arrangement of tasks depends on runtime data.

The dynamic-workflow module warns that generated workflows are processed like any other workflow and recommends keeping them under fifty tasks. For large-scale identical runs, it recommends map tasks.