Skip to main content

Tasks and task execution

Choose the task abstraction​

When a task represents a backend operation rather than a Python function—such as a connector call, SQL operation, sensor, webhook, or container execution—build it as a PythonTask subclass. PythonTask is the base for tasks with a Python-native Interface; its class docstring directs ordinary function-backed tasks to PythonFunctionTask instead. The normal @task decorator follows that function-backed path: it creates TaskMetadata, selects a plugin with TaskPlugins.find_pythontask_plugin, and instantiates the selected PythonFunctionTask (or asynchronous variant).

You need:

  • an installed flytekit environment;
  • a concrete task class that supplies execute, either directly or through a compatible mixin;
  • an explicit Interface describing the Python inputs and outputs; and
  • a unique task type and task name.

The following is the direct extension pattern used by tests/flytekit/unit/extend/test_connector.py. The connector mixin supplies the execute implementation, so OpenAITask does not define a Python function of its own:

from flytekit.core.base_task import PythonTask, kwtypes
from flytekit.core.interface import Interface
from flytekit.extend.backend.base_connector import (
    ConnectorRegistry,
    SyncConnectorExecutorMixin,
)


def test_openai_connector():
    ConnectorRegistry.register(MockOpenAIConnector(), override=True)

    class OpenAITask(SyncConnectorExecutorMixin, PythonTask):
        def __init__(self, **kwargs):
            super().__init__(
                task_type="openai",
                interface=Interface(inputs=kwtypes(a=int), outputs=kwtypes(o0=int)),
                **kwargs,
            )

    t = OpenAITask(task_config={}, name="openai task")
    res = t(a=1)
    assert res == 1

MockOpenAIConnector is the connector test fixture registered immediately before the task is called. The observable result is the integer returned by the connector through the normal task-call interface. For an ordinary Python function, use the @task API instead of manually wrapping the function in PythonTask.

Declare the Python-native interface​

PythonTask.__init__ accepts task_type, name, optional task_config, an Interface, an optional environment dictionary, and deck settings. It converts the Python interface with transform_interface_to_typed_interface(..., allow_partial_artifact_id_binding=True) for Flyte's typed task model, while retaining the original interface as python_interface for native conversion and execution.

The built-in Echo task in flytekit/core/task.py shows the complete construction pattern. It creates output names from the input count, passes a typed Interface to PythonTask, and implements the required execute method:

from typing import Any, Dict, Optional, Type

from flytekit.core.base_task import PythonTask
from flytekit.core.interface import Interface


class Echo(PythonTask):
    _TASK_TYPE = "echo"

    def __init__(self, name: str, inputs: Optional[Dict[str, Type]] = None, **kwargs):
        outputs = dict(zip(output_name_generator(len(inputs)), inputs.values())) if inputs else None
        super().__init__(
            task_type=self._TASK_TYPE,
            name=name,
            interface=Interface(inputs=inputs, outputs=outputs),
            **kwargs,
        )

    def execute(self, **kwargs) -> Any:
        values = list(kwargs.values())
        if len(values) == 1:
            return values[0]
        else:
            return tuple(values)

output_name_generator is the helper used by the actual Echo implementation to create output names. Echo is handled by FlytePropeller's echo plugin and does not create a pod; its source comments also require the echo plugin to be enabled in Propeller. The important PythonTask details are visible on the instance:

  • python_interface is the original native Interface.
  • interface is the transformed Flyte typed interface.
  • task_config is the plugin-specific configuration passed to the constructor.
  • environment returns the supplied Dict[str, str], or an empty dictionary when none was supplied.
  • get_type_for_input_var, get_type_for_output_var, and get_input_types read directly from the native interface.

Omitting interface is allowed by the constructor, but it creates an empty Interface. A concrete task that expects inputs or outputs should therefore pass its actual interface explicitly. Input names must also satisfy the rules enforced by Interface; they are not arbitrary display labels.

Compile a call into a workflow node​

A PythonTask call has two distinct meanings depending on the active Flyte context. During workflow compilation, PythonTask.compile delegates to create_and_link_node. The promise call handler then creates a Node whose metadata comes from entity.construct_node_metadata() and creates one Promise for each declared output. A task with no outputs receives a VoidPromise.

PythonTask.construct_node_metadata maps the task name, timeout, retry strategy, and interruptibility into workflow NodeMetadata:

return _workflow_model.NodeMetadata(
    name=extract_obj_name(self.name),
    timeout=self.metadata.timeout,
    retries=self.metadata.retry_strategy,
    interruptible=self.metadata.interruptible,
)

The returned promises are keyed by the names in the typed interface. This is why a task call inside a workflow can be used as an input to a downstream task without executing the task body while the workflow is being built.

Execute the task locally​

A local task call enters the promise handler in flytekit/core/promise.py. When the context is in local execution, the handler calls entity.local_execute(ctx, **kwargs); when the context is compiling, it calls create_and_link_node instead. Task.local_execute, inherited by PythonTask, performs the local boundary conversion:

  1. translate_inputs_to_literals converts native values and upstream Promise values into a LiteralMap using the task's typed and native input types.
  2. If local caching is enabled for the task, it looks up the literal map in LocalTaskCache.
  3. A cache miss or an uncached call enters sandbox_execute, which creates the task sandbox/checkpoint context and calls dispatch_execute.
  4. The resulting literals are checked against the declared output count.
  5. Each output is wrapped in a Promise; a zero-output task returns VoidPromise.

The promise handler also has an eager-local mode. In that mode it calls local_execute and converts the result back to native values with create_native_named_tuple; in the regular local path it returns promises.

You can verify the dispatch boundary directly with the same dictionary conversion test used in tests/flytekit/unit/core/test_type_hints.py:

import flytekit
from flytekit.core import context_manager
from flytekit.core.type_engine import TypeEngine
from flytekit import task


@task
def t2(a: dict) -> str:
    return ", ".join([f"K: {k} V: {v}" for k, v in a.items()])


task_spec = get_serializable(OrderedDict(), serialization_settings, t2)
input_map = {"a": {"k1": "v1", "k2": "2"}}
ctx = context_manager.FlyteContext.current_context()
lm = TypeEngine.dict_to_literal_map(ctx, d=input_map, type_hints={"a": dict})
output_lm = t2.dispatch_execute(ctx, lm)
str_value = output_lm.literals["o0"].scalar.primitive.string_value

The test inspects str_value after dispatch. PythonTask._literal_map_to_python_input is the corresponding implementation method; it calls TypeEngine.literal_map_to_kwargs(ctx, literal_map, self.python_interface.inputs) before invoking execute.

Follow the runtime dispatch lifecycle​

At the execution boundary, flytekit/bin/entrypoint.py::_dispatch_execute loads the task, downloads inputs.pb, converts the protobuf to Flytekit's LiteralMap model, and calls task_def.dispatch_execute(ctx, idl_input_literals). It then writes the returned output map, or writes an empty map for a VoidPromise:

task_def = load_task()
local_inputs_file = os.path.join(ctx.execution_state.working_dir, "inputs.pb")
ctx.file_access.get_data(inputs_path, local_inputs_file)
input_proto = utils.load_proto_from_file(_literals_pb2.LiteralMap, local_inputs_file)
idl_input_literals = _literal_models.LiteralMap.from_flyte_idl(input_proto)

outputs = task_def.dispatch_execute(ctx, idl_input_literals)

if isinstance(outputs, VoidPromise):
    logger.warning("Task produces no outputs")
    output_file_dict = {_constants.OUTPUT_FILE_NAME: _literal_models.LiteralMap(literals={})}
elif isinstance(outputs, _literal_models.LiteralMap):
    # The keys in this map hold the filenames to the offloaded proto literals.
    offloaded_literals: Dict[str, _literal_models.Literal] = {}
    literal_map_copy = {}

PythonTask.dispatch_execute runs the following lifecycle:

  1. It calls pre_execute(ctx.user_space_params) before converting any input literals.
  2. If decks are enabled, it adds the timeline deck when DeckField.TIMELINE is selected and builds user parameters with deck output enabled.
  3. It installs those returned user parameters into a child execution context while retaining the working directory.
  4. It converts Flyte literals to native Python inputs.
  5. It calls the concrete task's execute(**native_inputs).
  6. It calls post_execute(new_user_params, native_outputs).
  7. If the post-execution result is already a LiteralMap or DynamicJobSpec, it returns it unchanged. Otherwise it converts native outputs with _output_to_literal_map and writes configured decks.

The default hooks are intentionally small:

def pre_execute(self, user_params: Optional[ExecutionParameters]) -> Optional[ExecutionParameters]:
    return user_params


def post_execute(self, user_params: Optional[ExecutionParameters], rval: Any) -> Any:
    return rval

Override pre_execute when setup must occur before type transformers process inputs. Its docstring specifically calls out modifying user-space parameters and setting up services such as SparkSession before conversion. Override post_execute to clean up or change the return value after execute; it can also allow an IgnoreOutputs exception to propagate to the caller layer.

ShellTask demonstrates both hooks in a concrete subclass. It delegates pre_execute and post_execute to its configuration task, while its execute interpolates the script, runs it, and returns FlyteFile or FlyteDirectory outputs:

def pre_execute(self, user_params: ExecutionParameters) -> ExecutionParameters:
    return self._config_task_instance.pre_execute(user_params)


def post_execute(self, user_params: ExecutionParameters, rval: typing.Any) -> typing.Any:
    return self._config_task_instance.post_execute(user_params, rval)

If input conversion or user code fails, local and remote execution are deliberately reported differently. In local execution, input conversion errors and user exceptions are re-raised with the task name added to the message. During remote execution, input conversion and output conversion errors become FlyteNonRecoverableSystemException, while an exception from execute becomes FlyteUserRuntimeException. FlyteUploadDataException and FlyteDownloadDataException are allowed to propagate in both paths.

Return the declared output shape​

PythonTask._output_to_literal_map uses the declared output names and their order:

  • With zero declared outputs, the native result becomes an empty map.
  • With one output, a scalar is bound to that output. A one-element NamedTuple uses its tuple element when python_interface.output_tuple_name is present.
  • With multiple outputs, the native result is indexed in declaration order and mapped to the declared output names.

A plain tuple is rejected while converting an individual output. A one-output task therefore must not return an ordinary tuple merely because the value happens to contain one element. Task.local_execute also checks that the number of produced literals equals the number of declared outputs and raises an assertion if they differ.

Each output is converted asynchronously with TypeEngine.async_to_literal. If an output transformer fails, the raised error includes the task name, output position or user-defined output key, the received Python type, and the expected Python type. This gives you a task-specific failure at the literal boundary rather than returning an untyped Python value.

Output metadata is carried through the same conversion. When the active context has an OutputMetadataTracker, _output_to_literal_map attaches additional metadata to the resulting literal. Dynamic partitions and time partitions are serialized as an ArtifactID, encoded with base64, and stored under _uap (DYNAMIC_PARTITIONS). The artifact test verifies this behavior through dispatch_execute:

import datetime
from collections import OrderedDict
from typing import Annotated

import pandas as pd

from flytekit import Artifact, Inputs, task
from flytekit.core.context_manager import FlyteContextManager, OutputMetadataTracker
from flytekit.core.type_engine import TypeEngine


ctx = FlyteContextManager.current_context()
omt = OutputMetadataTracker()
ctx = ctx.with_output_metadata_tracker(omt).build()

a1_t_ab = Artifact(
    name="my_data", partition_keys=["a", "b"], time_partitioned=True, time_partition_granularity=Granularity.MONTH
)


@task
def t1(b_value: str, dt: datetime.datetime) -> Annotated[pd.DataFrame, a1_t_ab(b=Inputs.b_value)]:
    df = pd.DataFrame({"a": [1, 2, 3], "b": [b_value, b_value, b_value]})
    return a1_t_ab.create_from(df, a="dynamic!", time_partition=dt)


d = datetime.datetime(2021, 1, 1, 0, 0)
lm = TypeEngine.dict_to_literal_map(ctx, {"b_value": "my b value", "dt": d})
lm_outputs = t1.dispatch_execute(ctx, lm)
dyn_partition_encoded = lm_outputs.literals["o0"].metadata["_uap"]

The test decodes dyn_partition_encoded into an ArtifactID and verifies that partition a is dynamic! and that the time partition is January 1, 2021 with monthly granularity. The Artifact, Inputs, and Granularity names in this example are the same entities used by tests/flytekit/unit/core/test_artifacts.py; retain their corresponding imports when running the complete test.

Configure retries, timeouts, and task metadata​

TaskMetadata is the metadata companion shared by Task and PythonTask. Its fields include retries, timeout, interruptible, caching fields, deprecation text, pod_template_name, generates_deck, and is_eager. TaskMetadata.__post_init__ converts an integer timeout to datetime.timedelta(seconds=...) and validates related cache settings.

For example, the existing monitoring guide configures a function-backed task with the same metadata object that a PythonTask-derived implementation receives through **kwargs:

from flytekit import task, TaskMetadata
import datetime


@task(
    metadata=TaskMetadata(
        retries=3,
        timeout=datetime.timedelta(minutes=5),
        cache=True,
        cache_version="v1.0",
        cache_ignore_input_vars=("debug_mode",),
        interruptible=True,
    )
)
def long_running_task(input_data: str, debug_mode: bool = False) -> str:
    import random, time
    if random.random() < 0.1:
        raise ValueError("Simulated transient error")
    time.sleep(10)
    return f"Processed {input_data}"

For direct PythonTask subclasses, pass the same metadata through the constructor's keyword arguments. When a call is compiled, construct_node_metadata uses metadata.timeout, metadata.retry_strategy, and metadata.interruptible; retry_strategy is constructed from metadata.retries. When the task is serialized, TaskMetadata.to_taskmetadata_model maps caching to discoverable and discovery_version, and maps retries, timeout, interruptibility, cache-ignore variables, pod-template name, deck generation, deprecation, and eager execution to the Flyte task metadata model.

The validation is strict:

  • cache=True requires a non-empty cache_version when TaskMetadata is constructed.
  • cache_serialize=True requires cache=True.
  • cache_ignore_input_vars requires cache=True.
  • timeout must be an integer number of seconds or a datetime.timedelta.

The high-level @task wrapper also supports the newer Cache object. If cache=True is used without a legacy cache_version, the wrapper creates a Cache; mixing a Cache object with cache_serialize, cache_version, or cache_ignore_input_vars raises ValueError because those legacy arguments are deprecated in that path.

Verify local caching​

Caching is consulted by Task.local_execute, not by calling LocalTaskCache directly. It runs only when self.metadata.cache is true and LocalConfig.auto().cache_enabled is true. A cache lookup uses the task name, cache version, literal inputs, and ignored input names. On a miss, the task executes and the resulting literal map is stored; on a hit, the stored map is converted back into promises without executing the task body.

The cache key is assembled in flytekit/core/local_cache.py as:

return f"{task_name}-{cache_version}-{joblib.hash(hashed_inputs)}"

Before hashing, _calculate_cache_key removes every input named in cache_ignore_input_vars, replaces nested literals with their hash placements where available, and serializes the literal map deterministically. The persistent diskcache location is the hard-coded ~/.flyte/local-cache path in that file.

The local-cache test makes the behavior visible with an execution counter:

from flytekit import task, workflow


@task(cache=True, cache_version="v1")
def is_even(n: int) -> bool:
    global n_cached_task_calls
    n_cached_task_calls += 1
    return n % 2 == 0


@task(cache=False)
def uncached_task(a: int, b: int) -> int:
    return a + b


@workflow
def check_evenness(n: int) -> bool:
    uncached_task(a=n, b=n)
    return is_even(n=n)


assert n_cached_task_calls == 0
assert check_evenness(n=1) is False
assert n_cached_task_calls == 1
assert check_evenness(n=1) is False
assert n_cached_task_calls == 1
assert check_evenness(n=8) is True
assert n_cached_task_calls == 2

The test also verifies the environment controls:

monkeypatch.setenv("FLYTE_LOCAL_CACHE_ENABLED", "false")

assert is_even(n=1) is False
assert n_cached_task_calls == 1
assert is_even(n=1) is False
assert n_cached_task_calls == 2

FLYTE_LOCAL_CACHE_ENABLED=false causes repeated local calls to execute again. FLYTE_LOCAL_CACHE_OVERWRITE=true bypasses a cache hit, executes the task, and stores the new result; setting it back to false allows the new value to be read on the next call. Because the key contains the task name and cache version but not the function implementation, change the cache version when changing task logic if the old result is no longer valid.

Generate decks from execution values​

Deck generation is opt-in for a directly constructed PythonTask. If neither deck flag is passed, _disable_deck defaults to True. Passing enable_deck=True enables decks and retains the requested deck_fields; the constructor's default fields are SOURCE_CODE, DEPENDENCIES, TIMELINE, INPUT, and OUTPUT. Passing both enable_deck and disable_deck raises ValueError. disable_deck is deprecated and emits a FutureWarning; reading the deprecated disable_deck property emits a deprecation warning as well.

Use enable_deck in new code. The deck test verifies the context change only while the task runs:

from flytekit import task
from flytekit.core.context_manager import FlyteContextManager


@task(enable_deck=True)
def t1():
    ctx = FlyteContextManager.current_context()
    assert ctx.user_space_params.enable_deck == True
    return


ctx = FlyteContextManager.current_context()
assert ctx.user_space_params.enable_deck == False

t1()

ctx = FlyteContextManager.current_context()
assert ctx.user_space_params.enable_deck == False

During dispatch_execute, an enabled task adds the timeline deck to user_space_params.decks when DeckField.TIMELINE is selected, then builds user parameters with enable_deck=True. _write_decks renders selected input and output values with TypeEngine.to_html. For local execution it publishes the deck with _output_deck; remote execution leaves final deck output to the runtime entrypoint. deck_fields are discarded when decks are disabled, and every retained value must be a member of the DeckField enum.

Extend the pattern for backend tasks​

Once the typed interface and lifecycle are clear, choose the existing subclass pattern that matches the execution boundary:

  • ContainerTask and PythonAutoContainerTask call PythonTask with a typed interface, metadata, security context, task type, and environment. PythonAutoContainerTask initializes its container image before calling super().__init__ because Task.__init__ registers every task in FlyteEntities.entities and serialization can inspect the object during initialization.
  • SQLTask is a PythonTask subclass whose native execute deliberately raises NotImplementedError; SQL execution is delegated to its backend. Local tests use task_mock when they need to replace that execution.
  • BaseSensor, WebhookTask, ArrayNodeMapTask, and plugin task classes use the same PythonTask interface and dispatch boundary.
  • Connector implementations use SyncConnectorExecutorMixin or AsyncConnectorExecutorMixin to supply the executor behavior, as in the OpenAITask example.

IgnoreOutputs is a specific runtime control-flow exception, documented in flytekit/core/base_task.py for distributed training and peer-to-peer algorithms. The entrypoint recognizes it—including when it is wrapped in FlyteUserRuntimeException—and skips writing or uploading task outputs. It is not the general mechanism for hiding an ordinary execution failure.

Your final direct task should therefore have four concrete pieces: a unique task_type, a unique name, an explicit native Interface, and an execute implementation supplied by the task or an executor mixin. Use TaskMetadata for retries, timeouts, caching, and interruptibility; use enable_deck=True and selected DeckField values when HTML output is required; and test both task(...) calls for local execution and dispatch_execute(...) for the literal/runtime boundary.