# Tasks and workflows

The biggest structural change in Flyte 2 is that **everything is a task**. The `@task`, `@workflow`, and `@dynamic` decorators all collapse into a single `@env.task` on a `flyte.TaskEnvironment`, and a "workflow" is just a task that calls other tasks. This page covers the basic structure; see [Task configuration](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/tasks-and-workflows/configuration) for the environment settings and [Migration](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/tasks-and-workflows/overview) for the big picture.

## Hello world: tasks and workflows

A `@task` plus `@workflow` becomes two `@env.task`s, where the entrypoint task calls the others. Sequential calls are naturally ordered — no `>>` operator required.

### Flyte 1

```python
import flytekit

@flytekit.task
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@flytekit.task
def to_upper(greeting: str) -> str:
    return greeting.upper()

@flytekit.workflow
def main(name: str) -> str:
    greeting = say_hello(name=name)
    return to_upper(greeting=greeting)
```

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

### Flyte 2

```
import flyte

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

@env.task
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@env.task
def to_upper(greeting: str) -> str:
    return greeting.upper()

# The "workflow" is now just a task that calls other tasks.
@env.task
def main(name: str) -> str:
    greeting = say_hello(name)
    return to_upper(greeting)
```

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

## Chaining and ordering

In Flyte 1 you sometimes used `>>` to force ordering between tasks with no data dependency. In Flyte 2, sequential (synchronous) calls run in the order they're written, and `await`ing async tasks in sequence does the same. The `>>` operator is gone.

### Flyte 1

```python
from flytekit import task, workflow

@task
def clear_staging_table() -> None:
    # Side effect only: truncate the staging table.
    print("cleared staging table")

@task
def load_into_staging() -> None:
    # Side effect only: load fresh rows into staging.
    print("loaded staging table")

@task
def publish_to_prod() -> None:
    # Side effect only: swap staging into the production table.
    print("published to prod")

@workflow
def main() -> None:
    clear = clear_staging_table()
    load = load_into_staging()
    publish = publish_to_prod()

    # These tasks pass no data between them, so use the >> operator to force
    # ordering: clear must finish before load, which must finish before publish.
    clear >> load >> publish
```

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

### Flyte 2

```
import flyte

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

@env.task
def clear_staging_table() -> None:
    print("cleared staging table")

@env.task
def load_into_staging() -> None:
    print("loaded staging table")

@env.task
def publish_to_prod() -> None:
    print("published to prod")

# Sequential (synchronous) calls run in the order they're written, even when no
# data flows between them. The Flyte 1 `>>` ordering operator is gone.
@env.task
def main() -> None:
    clear_staging_table()
    load_into_staging()
    publish_to_prod()
```

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

## Subworkflows

A `@workflow` invoked by another `@workflow` (for example, a reusable preprocessing pipeline) becomes a task that calls other tasks — nest them as deeply as you like.

### Flyte 1

```python
from flytekit import task, workflow

@task
def impute(value: float) -> float:
    # Replace missing/negative sentinel values with 0.
    return value if value >= 0 else 0.0

@task
def scale(value: float) -> float:
    return value / 100.0

@workflow
def preprocess(value: float) -> float:
    imputed = impute(value=value)
    return scale(value=imputed)

@workflow
def main(raw_value: float) -> float:
    return preprocess(value=raw_value)
```

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

### Flyte 2

```
import flyte

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

@env.task
def impute(value: float) -> float:
    # Replace missing/negative sentinel values with 0.
    return value if value >= 0 else 0.0

@env.task
def scale(value: float) -> float:
    return value / 100.0

# A preprocessing "subworkflow" is just a task that calls other tasks.
@env.task
def preprocess(value: float) -> float:
    imputed = impute(value)
    return scale(imputed)

@env.task
def main(raw_value: float) -> float:
    return preprocess(raw_value)
```

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

## TaskEnvironment configuration

The `TaskEnvironment` holds the configuration that Flyte 1 spread across the `@task` decorator. The task decorator can still override a few settings per-task.

```python
import flyte

env = flyte.TaskEnvironment(
    name="my_env",                           # Required: unique name
    image=flyte.Image.from_debian_base(),    # Or a string, or "auto"
    resources=flyte.Resources(
        cpu="2",
        memory="4Gi",
        gpu="A100:1",
        disk="10Gi",
    ),
    env_vars={"LOG_LEVEL": "INFO"},
    secrets=[flyte.Secret(key="api-key", as_env_var="API_KEY")],
    cache="auto",                            # "auto", "override", "disable", or a Cache object
    reusable=flyte.ReusePolicy(replicas=5, idle_ttl=60),
    interruptible=True,
)

# The task decorator can override some settings:
@env.task(
    short_name="my_task",   # Display name
    cache="disable",        # Override cache
    retries=3,              # Retry count
    timeout=3600,           # Seconds or a timedelta
    report=True,            # Generate an HTML report
)
def my_task(x: int) -> int:
    return x
```

## Parameter mapping: `@task` → `TaskEnvironment` + `@env.task`

| Flyte 1 `@task` parameter | Flyte 2 location | Notes |
|---|---|---|
| `container_image` | `TaskEnvironment(image=...)` | Env-level only |
| `requests` | `TaskEnvironment(resources=...)` | Env-level only |
| `limits` | `TaskEnvironment(resources=...)` | Combined with requests (single value) |
| `environment` | `TaskEnvironment(env_vars=...)` | Env-level only |
| `secret_requests` | `TaskEnvironment(secrets=...)` | Env-level only |
| `cache` | Both | Can override at task level |
| `cache_version` | `flyte.Cache(version_override=...)` | In a `Cache` object |
| `retries` | `@env.task(retries=...)` | Task-level only |
| `timeout` | `@env.task(timeout=...)` | Task-level only |
| `interruptible` | Both | Can override at task level |
| `pod_template` | Both | Can override at task level |
| `deprecated` | N/A | Not in Flyte 2 |
| `docs` | `@env.task(docs=...)` | Task-level only |

For image, resource, secret, and caching detail, see [Task configuration](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/tasks-and-workflows/configuration).

## Next

- [Task configuration](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/tasks-and-workflows/configuration) — image, resources, caching, secrets, and scheduling
- [Control flow](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/tasks-and-workflows/control-flow) — conditionals, dynamic behavior, and error handling
- [Parallelism and fan-out](https://www.union.ai/docs/v2/union/user-guide/migration/flyte-2/tasks-and-workflows/parallelism) — running tasks in parallel

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