Control flow

Flyte 1 expressed branching, dynamic fan-out, and failure handling through DSL constructs (conditional(), @dynamic, @workflow(on_failure=...)). In Flyte 2 these are all ordinary Python, because orchestration runs as real Python at runtime. See Migration for the overall approach.

Conditional execution

The conditional() DSL becomes ordinary Python if / elif / else — for example, choosing a model based on dataset size.

Flyte 1Flyte 2
conditional_v1.py 
from flytekit import task, workflow, conditional

@task
def train_gradient_boosting(n_rows: int) -> str:
    return f"trained gradient boosting on {n_rows} rows"

@task
def train_logistic_regression(n_rows: int) -> str:
    return f"trained logistic regression on {n_rows} rows"

@workflow
def main(n_rows: int) -> str:
    # Pick the model based on dataset size.
    return (
        conditional("model_choice")
        .if_(n_rows > 10_000)
        .then(train_gradient_boosting(n_rows=n_rows))
        .else_()
        .then(train_logistic_regression(n_rows=n_rows))
    )
conditional_v2.py 
import flyte

env = flyte.TaskEnvironment(name="conditional")

@env.task
def train_gradient_boosting(n_rows: int) -> str:
    return f"trained gradient boosting on {n_rows} rows"

@env.task
def train_logistic_regression(n_rows: int) -> str:
    return f"trained logistic regression on {n_rows} rows"

# Branching is now ordinary Python control flow -- no conditional() DSL.
@env.task
def main(n_rows: int) -> str:
    if n_rows > 10_000:
        return train_gradient_boosting(n_rows)
    return train_logistic_regression(n_rows)

Dynamic workflows

@dynamic existed so a task could generate a variable number of subtask calls at runtime (e.g. one per data partition discovered at runtime). In Flyte 2 every task can do this natively, so @dynamic simply disappears — loop over runtime data in an ordinary @env.task.

Flyte 1Flyte 2
dynamic_v1.py 
from flytekit import task, workflow, dynamic

@task
def list_partitions(n: int) -> list[int]:
    return list(range(n))

@task
def process_partition(partition_id: int) -> int:
    # Aggregate one data partition.
    return partition_id * 2

@dynamic
def process_all(partitions: list[int]) -> list[int]:
    results = []
    for partition_id in partitions:
        results.append(process_partition(partition_id=partition_id))
    return results

@workflow
def main(n: int) -> list[int]:
    partitions = list_partitions(n=n)
    return process_all(partitions=partitions)
dynamic_v2.py 
import flyte

env = flyte.TaskEnvironment(name="dynamic")

@env.task
def process_partition(partition_id: int) -> int:
    # Aggregate one data partition.
    return partition_id * 2

# No @dynamic decorator needed: a plain task can loop over runtime data (e.g. a
# variable number of partitions discovered at runtime) and call other tasks.
@env.task
def main(n: int) -> list[int]:
    return [process_partition(partition_id) for partition_id in range(n)]

Error handling

Flyte 1’s @workflow(on_failure=...) handler becomes ordinary Python try / except — catch a failed training run, run cleanup, and recover or re-raise.

Flyte 1Flyte 2
error_handling_v1.py 
from flytekit import task, workflow

@task
def train_fold(max_depth: int) -> float:
    if max_depth <= 0:
        raise ValueError("max_depth must be positive")
    # Return validation accuracy for this hyperparameter.
    return 0.90 + 0.001 * max_depth

@task
def notify_failure() -> None:
    print("training run failed -- sending alert")

# The on_failure handler runs if any node in the workflow fails. There is no
# try/except inside a Flyte 1 workflow.
@workflow(on_failure=notify_failure)
def main(max_depth: int) -> float:
    return train_fold(max_depth=max_depth)
error_handling_v2.py 
import flyte

env = flyte.TaskEnvironment(name="error_handling")

@env.task
async def train_fold(max_depth: int) -> float:
    if max_depth <= 0:
        raise ValueError("max_depth must be positive")
    return 0.90 + 0.001 * max_depth

# Failure handling is ordinary Python try/except -- no on_failure handler.
@env.task
async def main(max_depth: int) -> float:
    try:
        return await train_fold(max_depth)
    except ValueError as e:
        print(f"invalid hyperparameter ({e}); falling back to a safe default")
        # Recover with a safe default instead of failing the whole run.
        return await train_fold(max_depth=6)

Flyte 2 also exposes typed errors, so you can catch a specific failure and retry with more resources — a common need for memory-hungry training jobs:

try:
    return await train_fold(sample_size)
except flyte.errors.OOMError:
    # Retry the same task with a larger memory request.
    return await train_fold.override(
        resources=flyte.Resources(memory="16Gi")
    )(sample_size)

See Error handling for more.

Next