Task authoring and execution

When you build workflows in Flyte, tasks are the fundamental atomic units of execution. Each task encapsulates a discrete step of computation with strongly typed inputs and outputs, isolated runtime dependencies, and declarative execution metadata.

In flytekit, tasks operate across two distinct lifecycles:

1. Compilation and serialization time: flytekit inspects your Python functions, type annotations, and configuration parameters, translating them into Flyte IDL (Interface Definition Language) protobuf specifications (TaskTemplate, TypedInterface, TaskMetadata).
2. Container execution time: The runtime entrypoint (pyflyte-execute) spins up inside a container, resolves the specific task entity via a task resolver, unmarshals literal inputs from remote storage, executes the task code, and serializes the outputs back to Flyte literals.

---

Task Abstraction Hierarchy

Flytekit structures tasks through an object-oriented inheritance model located across flytekit.core.base_task and flytekit.core.python_function_task:

                    ┌─────────────────────────┐
                    │      Task (Base)        │  FlyteIDL TaskTemplate mirror
                    └────────────┬────────────┘  (Untyped interface, dispatch_execute)
                                 │
                    ┌────────────▼────────────┐
                    │       PythonTask        │  Bridges Flyte IDL TypedInterface
                    └────────────┬────────────┘  with Python native Interface & TypeEngine
                                 │
                    ┌────────────▼────────────┐
                    │ PythonAutoContainerTask │  Generates container commands
                    └──────┬────────────┬─────┘  (pyflyte-execute with resolver)
                           │            │
         ┌─────────────────▼──┐      ┌──▼──────────────────┐
         │ PythonFunctionTask │      │ PythonInstanceTask  │
         └─────────┬──────────┘      └─────────────────────┘
                   │                 (Execute overridden directly on class)
       ┌───────────┴───────────┐
       │                       │
┌──────▼──────────────────┐ ┌──▼──────────────────────────┐
│ AsyncPythonFunctionTask │ │ EagerAsyncPythonFunctionTask │
└─────────────────────────┘ └─────────────────────────────┘

Core Abstraction Classes

• Task (flytekit.core.base_task.Task): The root class closest to the Flyte IDL TaskTemplate. It manages the untyped _interface_models.TypedInterface, TaskMetadata, SecurityContext, and documentation models. It defines the abstract entrypoints: dispatch_execute(), pre_execute(), and execute(), as well as local_execute() and sandbox_execute().
• PythonTask (flytekit.core.base_task.PythonTask): Generic base class for all tasks that possess a Python-native Interface. It holds the _python_interface and uses TypeEngine to marshal between Flyte LiteralMap representations and Python native kwargs (_literal_map_to_python_input() and _output_to_literal_map()). It also manages Flyte Deck generation (_write_decks()).
• PythonAutoContainerTask (flytekit.core.python_auto_container.PythonAutoContainerTask): Adds container serialization support. It generates the default pyflyte-execute CLI commands passed to the container at runtime.
• PythonFunctionTask (flytekit.core.python_function_task.PythonFunctionTask): Inspects a Python callable using transform_function_to_interface to extract input/output types and docstrings. It supports different execution modes: standard execution (ExecutionBehavior.DEFAULT), dynamic workflows (ExecutionBehavior.DYNAMIC), and eager execution (ExecutionBehavior.EAGER).
• PythonInstanceTask (flytekit.core.python_function_task.PythonInstanceTask): Abstract class for tasks that do not accept a user-supplied Python function body, but instead have a platform-defined execute() method implemented directly on the class.
• AsyncPythonFunctionTask & EagerAsyncPythonFunctionTask (flytekit.core.python_function_task): Specialized subclasses that support Python asynchronous coroutines (async def) and eager workflow execution contexts.

---

Authoring Tasks with @task

To define a task, decorate a standard Python function with @task from flytekit:

import typing
from datetime import timedelta
from flytekit import task, Resources

@task(
    cache=True,
    cache_version="1.0",
    retries=3,
    timeout=timedelta(minutes=10),
    interruptible=True,
    requests=Resources(cpu="2", mem="1Gi"),
    limits=Resources(cpu="4", mem="2Gi"),
    environment={"MODEL_STAGE": "production"},
)
def train_model(x: list[float], epochs: int) -> typing.NamedTuple("ModelOutput", accuracy=float, loss=float):
    accuracy = 0.95
    loss = 0.05
    return accuracy, loss

Typed Output Helpers: kwtypes

Flyte task interfaces require named output variables. When returning multiple outputs from a task, you can return a standard typing.NamedTuple or use Flytekit's kwtypes helper from flytekit.core.base_task:

from flytekit import task, kwtypes

@task
def compute_metrics(values: list[int]) -> kwtypes(mean=float, total=int):
    return sum(values) / len(values), sum(values)

kwtypes(**kwargs) returns an OrderedDict[str, Type] mapping output field names to their corresponding Python types.

Decorator Parameters and TaskMetadata

When you apply @task(...), flytekit constructs a TaskMetadata object (flytekit.core.base_task.TaskMetadata) and associates it with your task:

Parameter	Type	Internal Behavior / Constraints
cache	bool or Cache	Marks the task output as discoverable/memoized. When True, cache_version must also be provided.
cache_version	str	Version identifier for the cache key. Modifying this string invalidates previously cached outputs.
cache_serialize	bool	Enforces serial execution of concurrent identical task executions when caching is enabled. cache=True is required.
cache_ignore_input_vars	Tuple[str, ...]	List of input variable names excluded from the hash when computing the cache key. cache=True is required.
retries	int	Translates to RetryStrategy(self.retries). Specifies how many times Flyte will re-attempt the task on node failure.
timeout	timedelta or int	Maximum execution duration before termination. If an int is passed, TaskMetadata.__post_init__ converts it to timedelta(seconds=timeout).
interruptible	Optional[bool]	When True, allows scheduling on spot / preemptible instances with lower QoS guarantees.
requests / limits	Resources	Specifies resource requests and limits (CPU, memory, storage, GPU) assigned to the container definition.
enable_deck	bool	Enables generation of Flyte Deck HTML reports containing input/output metrics, dependencies, source code, and timeline info.
container_image	Union[str, ImageSpec]	Overrides the default container image for this task or references an ImageSpec builder definition.
docs	Documentation	Structured task documentation (short and long descriptions). Defaults to values parsed from the function docstring.

TaskMetadata.__post_init__ verifies caching invariants:

if self.cache and not self.cache_version:
    raise ValueError("Caching is enabled ``cache=True`` but ``cache_version`` is not set.")
if self.cache_serialize and not self.cache:
    raise ValueError("Cache serialize is enabled ``cache_serialize=True`` but ``cache`` is not enabled.")
if self.cache_ignore_input_vars and not self.cache:
    raise ValueError(
        f"Cache ignore input vars are specified ``cache_ignore_input_vars={self.cache_ignore_input_vars}`` but ``cache`` is not enabled."
    )

---

Function Validation Rules and Task Resolution

When Flyte executes tasks in remote containers, it must reconstruct (rehydrate) the exact Python object from container CLI arguments.

The Module-Level Requirement

The default resolver (DefaultTaskResolver in flytekit.core.python_auto_container) relies on importlib.import_module to load your task function by its module path and function name.

Because of this, nested, local, or inner functions cannot be used as @task functions with the default resolver:

# INVALID: Will raise a ValueError during task construction
def my_outer_function():
    @task
    def my_inner_task(x: int) -> int:
        return x + 1
    return my_inner_task

PythonFunctionTask.__init__ validates this condition:

if self._task_resolver is default_task_resolver:
    if (
        not istestfunction(func=task_function)
        and isnested(func=task_function)
        and not is_functools_wrapped_module_level(task_function)
    ):
        raise ValueError(
            "TaskFunction cannot be a nested/inner or local function. "
            "It should be accessible at a module level for Flyte to execute it..."
        )

Exceptions:

• Test functions inside modules starting with test_ are permitted to be nested.
• Decorators wrapping tasks must preserve function metadata using functools.wraps or functools.update_wrapper.

How TaskResolverMixin Works

All container-backed tasks use a resolver that implements TaskResolverMixin (flytekit.core.base_task.TaskResolverMixin):

class TaskResolverMixin(object):
    @property
    @abstractmethod
    def location(self) -> str: ...

    @abstractmethod
    def loader_args(self, settings: SerializationSettings, t: Task) -> List[str]: ...

    @abstractmethod
    def load_task(self, loader_args: List[str]) -> Task: ...

At serialization time, PythonAutoContainerTask.get_default_command() constructs the container command:

pyflyte-execute \
  --inputs {{.input}} \
  --output-prefix {{.outputPrefix}} \
  --raw-output-data-prefix {{.rawOutputDataPrefix}} \
  --resolver flytekit.core.python_auto_container.default_task_resolver \
  -- \
  task-module my_project.workflows task-name train_model

At runtime, pyflyte-execute imports the resolver class specified by --resolver, passes the trailing arguments to resolver.load_task(loader_args), and executes the rehydrated Task instance.

---

Execution Modes and Variations

PythonFunctionTask supports distinct runtime patterns by altering its execution_mode:

1. Default Task Execution (ExecutionBehavior.DEFAULT)

Standard tasks run as standalone single-container executions. When called directly, execute(**kwargs) delegates straight to your decorated function:

if self.execution_mode == self.ExecutionBehavior.DEFAULT:
    return self._task_function(**kwargs)

2. Dynamic Tasks (ExecutionBehavior.DYNAMIC)

Created via the @dynamic decorator (flytekit.core.dynamic_workflow_task.dynamic), dynamic tasks allow runtime DAG creation based on input data:

from flytekit import task, dynamic

@task
def process_item(val: int) -> int:
    return val * 2

@dynamic
def dynamic_pipeline(count: int) -> list[int]:
    results = []
    for i in range(count):
        results.append(process_item(val=i))
    return results

How dynamic execution works internally:

• Local execution: dynamic_execute() converts inputs to native literals, creates a cached PythonFunctionWorkflow, runs it in Mode.LOCAL_DYNAMIC_TASK_EXECUTION, and translates outputs to a LiteralMap.
• Remote execution: Under Mode.TASK_EXECUTION, compile_into_workflow() compiles the dynamic function into an internal WorkflowSpec, serializes nested tasks and node dependencies, and returns a DynamicJobSpec (a dynamic workflow DAG returned to FlytePropeller).

3. Async and Eager Tasks (AsyncPythonFunctionTask & EagerAsyncPythonFunctionTask)

Flyte supports native Python async task definitions and eager workflows:

from flytekit import task, eager

@task
async def fetch_data(url: str) -> str:
    # Asynchronous task execution
    return f"data from {url}"

@eager
async def eager_workflow(url: str) -> str:
    data = await fetch_data(url=url)
    return data

• AsyncPythonFunctionTask executes asynchronous tasks using loop_manager.synced(async_execute). Eager and dynamic execution modes are mutually exclusive (EagerAsyncPythonFunctionTask raises NotImplementedError if combined with dynamic mode).
• When running on a remote cluster, EagerAsyncPythonFunctionTask uses a Controller worker queue to submit sub-task executions directly to FlyteAdmin, awaiting remote job completion from inside the task container.

4. Distributed Array Tasks (ArrayNodeMapTask)

For parallel processing over lists of inputs, Flyte provides map_task / ArrayNodeMapTask (flytekit.core.array_node_map_task):

from flytekit import task, map_task

@task
def double(val: int) -> int:
    return val * 2

mapped_double = map_task(double, concurrency=5, min_success_ratio=0.8)

ArrayNodeMapTask wraps a PythonFunctionTask (which must use ExecutionBehavior.DEFAULT and possess exactly one output variable) and configures Flyte's array node execution mechanism.

---

Local vs. Remote Execution Lifecycle

When a task executes, flytekit coordinates context management, input conversion, execution, and output conversion through PythonTask.dispatch_execute():

                  ┌───────────────────────────────┐
                  │ Input LiteralMap (from Flyte) │
                  └───────────────┬───────────────┘
                                  │
                                  ▼
               ┌─────────────────────────────────────┐
               │  pre_execute(user_space_params)     │
               └──────────────────┬──────────────────┘
                                  │
                                  ▼
      ┌────────────────────────────────────────────────────────┐
      │ _literal_map_to_python_input(input_literal_map, ctx)   │
      │ (TypeEngine converts Literals -> Python native kwargs) │
      └───────────────────────────┬────────────────────────────┘
                                  │
                                  ▼
               ┌─────────────────────────────────────┐
               │       execute(**native_inputs)      │
               │   (Executes user-defined function)  │
               └──────────────────┬──────────────────┘
                                  │
                                  ▼
               ┌─────────────────────────────────────┐
               │ post_execute(user_params, outputs)  │
               └──────────────────┬──────────────────┘
                                  │
                                  ▼
      ┌────────────────────────────────────────────────────────┐
      │ _output_to_literal_map(native_outputs, exec_ctx)       │
      │ (TypeEngine converts Python values -> LiteralMap)      │
      └───────────────────────────┬────────────────────────────┘
                                  │
                                  ▼
               ┌─────────────────────────────────────┐
               │ _write_decks(...)                   │
               │ (Renders HTML decks for task stats) │
               └──────────────────┬──────────────────┘
                                  │
                                  ▼
                  ┌───────────────────────────────┐
                  │ Output LiteralMap (to Flyte)  │
                  └───────────────────────────────┘

Pre-Execute and Post-Execute Hooks

• pre_execute(user_params: ExecutionParameters): Runs before any input literal translation. Plugins (such as Spark or custom GPU setups) override pre_execute() to configure environment parameters (like initializing a SparkSession) before type transformers run.
• post_execute(user_params, rval): Runs after the task function finishes. If a task raises IgnoreOutputs (flytekit.core.base_task.IgnoreOutputs), the output translation is bypassed (useful in distributed training contexts where secondary ranks produce no workflow artifacts).

Exception Handling Differences

PythonTask.dispatch_execute() handles exceptions differently based on whether execution is local or remote (ctx.execution_state.is_local_execution()):

• Local execution: The original Python exception is preserved and re-raised with augmented context so tracebacks remain clear during local debugging and testing:

if is_local_execution:
    e.args = (f"Error encountered while executing '{self.name}':\n  {e}",)
    raise

• Remote execution: User exceptions are caught and wrapped in FlyteUserRuntimeException(e), while framework or serialization errors are wrapped in FlyteNonRecoverableSystemException(e). This distinction signals FlytePropeller whether the failure is retryable or user-code-induced.

---

Local Execution and Caching

You can call any task locally as a regular Python function:

from flytekit import task

@task
def add(a: int, b: int) -> int:
    return a + b

# Invoked directly in Python
result = add(a=10, b=20)
assert result == 30

When called directly, Task.__call__ delegates to flyte_entity_call_handler.

When cache=True is configured on the task metadata, local execution checks the local task cache (LocalTaskCache via LocalConfig.auto()):

1. Flytekit computes a cache key using task.name, task.metadata.cache_version, and input literals (excluding any keys listed in cache_ignore_input_vars).
2. If LocalTaskCache.get(...) finds cached outputs, sandbox_execute() is bypassed and the cached LiteralMap is returned.
3. If a cache miss occurs, sandbox_execute() runs the task and stores the output in LocalTaskCache.set(...).
4. If local_config.cache_overwrite is set to True, cached outputs are ignored and the task re-executes.