The Task Model and Execution Lifecycle

What you will build

You will trace one Flyte task from its Python declaration to its runtime representation, then verify the same lifecycle with a container-backed task. The path uses TaskMetadata for execution policy, PythonTask for a Python-native interface, Task.local_execute for local calls, and PythonTask.dispatch_execute for the input/execute/output pipeline. You will also see how PythonAutoContainerTask turns a task into a pyflyte-execute container command.

The examples use the same public decorator and task classes exercised by flytekit's tests. They assume flytekit's test/runtime dependencies are installed and, for container serialization, that you have a usable SerializationSettings.image_config.

1. Start with the public task declaration

The normal user-facing entry point is @task. The decorator accepts ordinary Python annotations and can also receive task-specific configuration and retries:

from typing import Dict

from flytekit import task


@task
def my_task(x: int, y: Dict[str, str]) -> str:
    ...


@task(task_config=Spark(), retries=3)
def my_task(x: int, y: Dict[str, str]) -> str:
    ...

The first declaration is the plain Python path; the second is the task-configured form shown in flytekit/core/task.py. In the usual Python-function path, flytekit creates a PythonFunctionTask, which is a Python-native, auto-container task. A task call is not an immediate ordinary Python function call in every context: Task.__call__ delegates to flyte_entity_call_handler, which chooses the context-appropriate behavior.

2. Attach execution policy with TaskMetadata

Use TaskMetadata when you need caching, retries, timeouts, or other task-level policy. The decorator forwards options such as cache, cache_version, and retries into this metadata; lower-level task wrappers can construct it directly.

from flytekit import task


@task(cache=True, cache_serialize=True, cache_version="1.0")
def foo(i: str):
    print(f"{i}")


foo_metadata = foo.metadata
assert foo_metadata.cache is True
assert foo_metadata.cache_serialize is True
assert foo_metadata.cache_version == "1.0"

TaskMetadata.__post_init__ validates the combinations. cache=True requires a non-empty cache_version; cache_serialize=True and cache_ignore_input_vars require caching to be enabled. An integer timeout is interpreted as seconds and converted to a datetime.timedelta. The following invalid declarations therefore raise ValueError, as covered by tests/flytekit/unit/core/test_python_function_task.py:

from flytekit import task


with pytest.raises(ValueError):
    @task(cache=True)
    def foo_missing_cache_version(i: str):
        print(f"{i}")


with pytest.raises(ValueError):
    @task(cache_serialize=True)
    def foo_missing_cache(i: str):
        print(f"{i}")

For registration, TaskMetadata.to_taskmetadata_model() creates the Flyte task metadata model. It maps cache to discoverable, carries the cache version and ignored input names, converts retries through retry_strategy, and includes timeout, interruptibility, deprecation text, pod-template name, deck generation, and eager-execution settings. retry_strategy itself is a Flyte literal-model retry strategy built from retries.

3. Distinguish the two task interfaces

Task is the IDL-oriented base abstraction. Its constructor stores a task type, name, Flyte TypedInterface, metadata, task-type version, security context, and documentation, then appends the task to FlyteEntities.entities. Its interface property is therefore the Flyte-typed interface used for literal maps and serialization.

Task does not provide a Python-native interface: python_interface returns None, and get_input_types() returns None. It is an infrastructure base and cannot be used as a ready-to-run task because compile, dispatch_execute, pre_execute, and execute are abstract or unimplemented.

PythonTask adds the Python side. Its constructor accepts an Interface, transforms it with transform_interface_to_typed_interface, and retains the original interface as python_interface. Its type lookups use the declared Python types by variable name:

python_type = task.get_type_for_input_var("input_name", value)
output_type = task.get_type_for_output_var("output_name", value)
input_types = task.get_input_types()

The concrete PythonTask implementation returns the corresponding entries from python_interface.inputs and python_interface.outputs; get_input_types() returns the complete Python input mapping. This gives the local conversion path both sides of the boundary:

• Task.interface describes Flyte variables and literal types.
• PythonTask.python_interface describes the native Python inputs and outputs.

4. Observe the invocation branch

A task call can become a workflow node or a local result. During workflow compilation, PythonTask.compile calls create_and_link_node, so the call contributes a node to the workflow rather than running user code immediately. The container-task test demonstrates both forms with the same task:

from flytekit import task, workflow
from flytekit.core.container_task import ContainerTask
from flytekit.core.type_engine import kwtypes


calculate_ellipse_area_python_template_style = ContainerTask(
    name="calculate_ellipse_area_python_template_style",
    input_data_dir="/var/inputs",
    output_data_dir="/var/outputs",
    inputs=kwtypes(a=float, b=float),
    outputs=kwtypes(area=float, metadata=str),
    image="ghcr.io/flyteorg/rawcontainers-python:v2",
    command=[
        "python",
        "calculate-ellipse-area.py",
        "{{.inputs.a}}",
        "{{.inputs.b}}",
        "/var/outputs",
    ],
)


area, metadata = calculate_ellipse_area_python_template_style(a=3.0, b=4.0)
assert isinstance(area, float)
assert isinstance(metadata, str)


@task
def t1(a: float, b: float):
    return a + b, a * b


@workflow
def wf(a: float, b: float):
    a, b = t1(a=a, b=b)
    area, metadata = calculate_ellipse_area_python_template_style(a=a, b=b)
    return area, metadata


area, metadata = wf(a=3.0, b=4.0)
assert isinstance(area, float)
assert isinstance(metadata, str)

The first call follows local execution in the test. The call inside wf is compiled into workflow nodes. For a PythonTask, compile returns the promises created by create_and_link_node; for local execution, Task.local_execute performs the conversion and dispatch described next.

5. Follow local execution into dispatch

When you invoke a task locally with native values or promises, Task.local_execute first calls translate_inputs_to_literals. It uses the task's Flyte input variables and, for a PythonTask, the native input-type mapping to build a LiteralMap. Conversion failures are annotated with the task name.

If metadata caching is enabled, local execution consults LocalConfig.auto(). The cache is used only when both metadata.cache and local_config.cache_enabled are true. LocalConfig.cache_overwrite bypasses a hit and executes the task. Otherwise flytekit looks up LocalTaskCache using the task name, cache version, input literal map, and cache_ignore_input_vars; a cache miss dispatches the task and stores the resulting literal map.

Whether the result came from cache or execution, local_execute checks that the number of returned literals equals the number of declared outputs. A task with no declared outputs returns VoidPromise(self.name). Otherwise it creates a Promise for each declared output and passes those promises to create_task_output, using the Python interface when one is available.

The actual runtime pipeline is PythonTask.dispatch_execute. Its order is important:

1. Call pre_execute(ctx.user_space_params) before converting input literals.
2. Install the returned user parameters in a derived FlyteContext while retaining the working directory.
3. Convert the input LiteralMap with TypeEngine.literal_map_to_kwargs and the Python input interface.
4. Call execute(**native_inputs).
5. Call post_execute(new_user_params, native_outputs).
6. Return an existing LiteralMap or DynamicJobSpec unchanged.
7. Convert ordinary native outputs with TypeEngine.async_to_literal.
8. Write enabled decks and return the output literal map.

The base hooks show where an extension can participate:

class PythonTask(...):
    def pre_execute(self, user_params):
        return user_params

    def execute(self, **kwargs):
        pass

    def post_execute(self, user_params, rval):
        return rval

execute is abstract in PythonTask; pre_execute and post_execute are no-ops unless overridden. The source documentation for pre_execute specifically places it before input conversion, making it the hook for changing user parameters or preparing resources needed by type transformers. post_execute receives the modified parameters and the return value, and can clean up or alter outputs.

For local execution, conversion and user exceptions retain the original exception and add task context. In hosted execution, the dispatch path wraps user failures in Flyte user/runtime exceptions and conversion failures in the corresponding non-recoverable system exceptions.

6. Check output conversion rules

PythonTask._output_to_literal_map maps Python return values to the declared output names before invoking the type engine. Multiple outputs are positional. A zero-output task produces an empty literal map. A single output normally maps the complete return value to that output; a single-output NamedTuple uses the first tuple element when output_tuple_name is present.

An individual output that is itself a tuple is rejected with TypeError. Successful values are converted asynchronously through TypeEngine.async_to_literal using both the Python type and the declared Flyte literal type. Output conversion errors include the task name and output position. Output metadata tracked in the execution context is attached to the resulting literal when present.

The dispatch path does not force every return value through this conversion. If execute or post_execute returns a Flyte LiteralMap or a dynamic DynamicJobSpec, dispatch_execute returns it directly. This is the short-circuit used by dynamic-task behavior and already-constructed literal results.

IgnoreOutputs is a marker exception for a successful execution whose outputs can be ignored. flytekit documents distributed training and peer-to-peer algorithms as its use cases. The runtime entrypoint treats FlyteUserRuntimeException(IgnoreOutputs()) specially and does not write an output file:

from unittest import mock

from flytekit import IgnoreOutputs
from flytekit.core import context_manager
from flytekit.core.exceptions import FlyteUserRuntimeException


python_task = mock.MagicMock()
python_task.dispatch_execute.side_effect = FlyteUserRuntimeException(IgnoreOutputs())

The entrypoint test verifies that dispatching this task leaves mock_write_to_file.call_count == 0. The PyTorch plugin uses the same contract: rank zero returns the task result, while a worker group without rank zero raises IgnoreOutputs().

7. Put user code in an auto-generated container

Use PythonAutoContainerTask as the base for an extension whose user code runs in the generated container. It extends PythonTask with an image or ImageSpec, Resources/ResourceSpec, environment, secrets, resolver, pod template, accelerator, and shared-memory settings. A minimal concrete subclass from tests/flytekit/unit/core/test_python_auto_container.py is:

from typing import Any

from flytekit.core.python_auto_container import PythonAutoContainerTask


class DummyAutoContainerTask(PythonAutoContainerTask):
    def execute(self, **kwargs) -> Any:
        pass


task = DummyAutoContainerTask(name="x", task_config=None, task_type="t")

This construction is verifiable by inspecting task.task_type, task.name, and the generated container through get_container(settings). For serialization, SerializationSettings.image_config must resolve the configured image or ImageSpec; the test fixture supplies an ImageConfig and serialization environment:

from flytekit.configuration import Image, ImageConfig, SerializationSettings


def default_serialization_settings():
    default_image = Image(name="default", fqn="docker.io/xyz", tag="some-git-hash")
    default_image_config = ImageConfig(default_image=default_image)
    return SerializationSettings(
        project="p", domain="d", version="v", image_config=default_image_config, env={"FOO": "bar"}
    )

PythonAutoContainerTask.get_default_command emits the command used by hosted Flyte. For the module-level task in the test, the exact result is:

[
    "pyflyte-execute",
    "--inputs", "{{.input}}",
    "--output-prefix", "{{.outputPrefix}}",
    "--raw-output-data-prefix", "{{.rawOutputDataPrefix}}",
    "--checkpoint-path", "{{.checkpointOutputPrefix}}",
    "--prev-checkpoint", "{{.prevCheckpointPrefix}}",
    "--resolver", "flytekit.core.python_auto_container.default_task_resolver",
    "--",
    "task-module", "tests.flytekit.unit.core.test_python_auto_container",
    "task-name", "task",
]

The resolver location and loader arguments are part of the serialized command. The default resolver reimports the module and looks up the named task, so module-level task objects can be rehydrated. Tasks that cannot be found by module-and-attribute lookup need a suitable TaskResolverMixin implementation; a resolver from the active compilation context takes precedence over a resolver passed to the constructor.

get_container(settings) resolves the registerable image, merges settings.env first with the task's environment second, and uses the generated command as container arguments. Thus task-level environment values overwrite same-name serialization settings values. If an ImageSpec has runtime_packages, serialization adds _F_RUNTIME_PACKAGES to the environment for the entrypoint to install before execution.

If you provide a pod_template, get_container() returns None; get_k8s_pod() returns the pod template with the generated container merged into it. The explicit pod_template_name constructor argument overwrites metadata.pod_template_name, including by clearing it when the argument is None. Do not combine the resources argument with requests or limits: the constructor raises ValueError. Secret requests must contain Secret objects.

Complete lifecycle result and next steps

At the end of this path, you can identify each boundary:

• Task stores the Flyte-facing identity and typed interface and registers the entity.
• TaskMetadata validates and serializes cache, retry, timeout, and execution policy.
• PythonTask pairs the Flyte interface with native Python types and owns dispatch hooks.
• local_execute translates native values or promises, optionally uses LocalTaskCache, dispatches, and returns promises or a VoidPromise.
• dispatch_execute runs pre_execute, input conversion, execute, post_execute, output conversion, and deck writing in that order.
• PythonAutoContainerTask adds the image, resources, environment, pod, and resolver data needed to run the task in a hosted container.

To continue, inspect PythonFunctionTask for the normal typed-function implementation behind @task, ContainerTask for a task with an explicitly templated command, and flytekit.bin.entrypoint for the runtime rehydration and output-publication boundary.