# 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](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/control-flow/overview) 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 1

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

*Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/migration/flyte-2/conditional_v1.py*

### Flyte 2

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

*Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/migration/flyte-2/conditional_v2.py*

## 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 1

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

*Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/migration/flyte-2/dynamic_v1.py*

### Flyte 2

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

*Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/migration/flyte-2/dynamic_v2.py*

## 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 1

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

*Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/migration/flyte-2/error_handling_v1.py*

### Flyte 2

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

*Source: https://github.com/unionai/unionai-examples/blob/main/v2/user-guide/migration/flyte-2/error_handling_v2.py*

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:

```python
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](https://www.union.ai/docs/v2/union/user-guide/migration/task-programming/error-handling) for more.

## Next

- [Parallelism and fan-out](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/control-flow/parallelism) — running many tasks in parallel
- [ML workloads](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/control-flow/ml-workloads) — training, HPO, and inference

---
**Source**: https://github.com/unionai/unionai-docs/blob/main/content/user-guide/migration/flyte-2/control-flow.md
**HTML**: https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/control-flow/