Skip to main content

Task authoring and execution

Declaring a task with @task​

When a Python function should become a Flyte task, annotate its inputs and return value and decorate it with @task. The decorator is the normal construction path for a PythonFunctionTask; the annotations become the task's typed interface rather than being used only by a static type checker.

from flytekit import task


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

For a plugin-specific task, pass its configuration and task options to the decorator:

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

The examples above are in the task decorator's source documentation. task.task builds TaskMetadata, selects the task implementation for the supplied configuration, and returns a task object while preserving the function wrapper. Coroutine functions are routed to AsyncPythonFunctionTask; ordinary annotated functions use PythonFunctionTask.

The task abstraction ladder​

Flytekit separates the Flyte task model from Python-specific invocation:

  • Task in base_task.py is the IDL-oriented base. It stores the task type, name, typed Flyte interface, metadata, task-type version, security context, and documentation. Constructing a task also registers it in FlyteEntities.entities.
  • PythonTask extends Task with a Python-native Interface, Python input/output types, workflow-node compilation, and the literal/native conversion pipeline. It is the extension base for tasks that have Python interfaces but do not necessarily execute a user function.
  • PythonFunctionTask extends the Python auto-container task path and binds PythonTask.execute to a user callable. It derives its interface with transform_function_to_interface, reads the function's docstring through Docstring, removes any ignore_input_vars, and derives the default name from the function's module.
  • PythonInstanceTask is the alternative for a platform-defined implementation. It is an abstract base for tasks without a user-defined function body; a subclass supplies execute, while the instance remains addressable by its module and variable name for task loading.

You normally do not instantiate Task directly: dispatch_execute, pre_execute, and execute are abstract there. PythonTask supplies the practical dispatch implementation, and PythonFunctionTask supplies the user-function implementation.

Configure task metadata​

Use task options on @task, or construct TaskMetadata when a task wrapper accepts metadata directly. The metadata object carries retries, timeout, cache settings, interruptibility, deprecation text, pod-template name, deck generation, and eager state into Flyte's task model.

from flytekit.core.base_task import TaskMetadata

metadata = TaskMetadata(
    retries=2,
    timeout=300,
    interruptible=True,
)

TaskMetadata.__post_init__ interprets an integer timeout as seconds and replaces it with a datetime.timedelta. It rejects a non-empty timeout that is neither an integer nor a datetime.timedelta. Caching has explicit validation:

TaskMetadata(cache=True, cache_version="v1")

cache=True without a non-empty cache_version raises ValueError. cache_serialize=True and cache_ignore_input_vars=("x",) are also rejected unless cache=True. The decorator's current API prefers a Cache object; the older cache arguments are deprecated, and the decorator rejects mixing a Cache object with those old arguments.

TaskMetadata.to_taskmetadata_model() converts these values to Flyte's task model. It wraps retries in a RetryStrategy, records the Flytekit SDK runtime and version, and passes through timeout, interruptibility, cache version and ignored inputs, deprecation text, deck generation, pod-template, and eager fields.

Python-task options​

PythonTask accepts a Python Interface, environment, and deck settings in addition to the common task arguments. Its constructor converts the Python interface into a typed Flyte interface and stores the task configuration for plugin-specific handling. If the interface contains a parsed docstring, it populates or updates the task's Documentation with the short and long descriptions.

Deck output is disabled by default. Choose exactly one of enable_deck or the deprecated disable_deck; passing both raises ValueError. When decks are enabled, deck_fields is checked against DeckField, and the selected fields are used to write input/output decks. PythonFunctionTask additionally writes source-code and dependency decks before delegating to PythonTask's deck writer.

PythonAutoContainerTask, the parent used by PythonFunctionTask, exposes container_image, resources, environment, and a task resolver. During serialization, the configured environment is combined with serialization settings and an explicit image or ImageSpec is resolved against the configured image defaults.

What happens when you call a task?​

A task call is not a direct call to the original Python function. Task.__call__ delegates to flyte_entity_call_handler, which chooses behavior from the current Flyte execution context. In workflow compilation, the call contributes a node. In local execution, it runs the task and returns native-looking task results backed by Promise objects.

The local path in Task.local_execute is:

native values / Promises
        │
        ▼
translate_inputs_to_literals
        │
        ▼
LiteralMap ── optional LocalTaskCache lookup
        │
        ▼
sandbox_execute → dispatch_execute
        │
        ▼
output LiteralMap
        │
        ▼
Promise(s) or VoidPromise

Task.local_execute translates native values, promises, lists, and dictionaries into literals using the task interface. If both TaskMetadata.cache and LocalConfig.auto().cache_enabled are true, it checks LocalTaskCache using the task name, cache version, input literal map, and configured ignored inputs. cache_overwrite bypasses a hit. A miss executes through sandbox_execute, then stores the resulting literal map.

The method 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 Promise objects and packages them with create_task_output.

The Python dispatch pipeline​

PythonTask.dispatch_execute performs the conversion and execution work for both local and hosted execution:

new_user_params = self.pre_execute(ctx.user_space_params)

native_inputs = self._literal_map_to_python_input(input_literal_map, exec_ctx)
with timeit("Execute user level code"):
    native_outputs = self.execute(**native_inputs)
native_outputs = self.post_execute(new_user_params, native_outputs)

if isinstance(native_outputs, (_literal_models.LiteralMap, _dynamic_job.DynamicJobSpec)):
    return native_outputs

literals_map, native_outputs_as_map = run_sync(
    self._output_to_literal_map, native_outputs, exec_ctx
)
return literals_map

The implementation in base_task.py first calls pre_execute, allowing a subclass to modify execution parameters before input conversion. _literal_map_to_python_input uses TypeEngine.literal_map_to_kwargs and the task's Python input types. execute then receives ordinary Python keyword arguments. post_execute can clean up or alter the result; its default implementation returns the result unchanged.

For ordinary outputs, _output_to_literal_map maps declared output names to returned values and calls TypeEngine.async_to_literal for each value. It handles zero outputs and the special one-output NamedTuple convention. A tuple used as an individual declared output raises TypeError. If a result is already a LiteralMap or a DynamicJobSpec, dispatch_execute returns it without attempting ordinary output conversion.

Local and hosted failures are handled differently. During local execution, input-conversion and user-function exceptions are re-raised with task context in their messages. During remote execution, user-code failures become FlyteUserRuntimeException, while conversion failures become FlyteNonRecoverableSystemException. IgnoreOutputs is a marker exception for flows such as distributed or peer-to-peer algorithms; the dispatch documentation says it is bubbled to the caller layer rather than caught by PythonTask.

Unit-test task calls​

To replace a Python task's execute method temporarily, use task_mock from testing.py:

from flytekit import task
from flytekit.core.testing import task_mock


@task
def t1(i: int) -> int:
    pass


with task_mock(t1) as m:
    m.side_effect = lambda x: x
    t1(10)
    # The mock is valid only within this context

The context manager accepts a PythonTask (as well as workflow and reference entities), installs a MagicMock around execute, and restores the original method when the context exits.

Serialization and task rehydration​

A hosted execution starts from a serialized task template, so the worker needs a way to reconstruct the Python task object. TaskResolverMixin defines that contract through location, name, loader_args(settings, task), load_task(loader_args), and get_all_tasks; task_name can optionally provide a custom serialized name.

PythonAutoContainerTask.get_default_command places the resolver location and loader arguments in the container command:

container_args = [
    "pyflyte-execute",
    "--inputs", "{{.input}}",
    "--output-prefix", "{{.outputPrefix}}",
    "--raw-output-data-prefix", "{{.rawOutputDataPrefix}}",
    "--checkpoint-path", "{{.checkpointOutputPrefix}}",
    "--prev-checkpoint", "{{.prevCheckpointPrefix}}",
    "--resolver", self.task_resolver.location,
    "--", *self.task_resolver.loader_args(settings, self),
]

The default resolver uses the task's module and name: it imports the module and looks up the task attribute. This is why a normal PythonFunctionTask must be accessible at module level. With the default resolver, PythonFunctionTask rejects nested or local functions unless they are test functions or a module-level function has been correctly wrapped with functools.wraps or functools.update_wrapper.

A custom resolver can use another identifier scheme. For example, ClassStorageTaskResolver stores tasks in an in-memory mapping and serializes the mapping index:

def load_task(self, loader_args: List[str]) -> PythonAutoContainerTask:
    if len(loader_args) != 1:
        raise RuntimeError(...)
    idx = int(loader_args[0])
    return self.mapping[idx]

def loader_args(self, settings, t):
    if t not in self.mapping:
        raise ValueError("no such task")
    return [f"{self.mapping.index(t)}"]

The two methods must be inverse operations for the serialized task to be rehydrated. A compilation context can override an explicitly supplied resolver; PythonAutoContainerTask uses compilation_state.task_resolver when present, otherwise the supplied resolver or default_task_resolver.

Dynamic, asynchronous, and eager tasks​

Dynamic tasks​

Use dynamic, which is defined as task.task partially applied with PythonFunctionTask.ExecutionBehavior.DYNAMIC, when the function body should use native Python control flow to create a runtime workflow:

from flytekit import dynamic


@dynamic
def my_dynamic_subwf(a: int) -> (list[str], int):
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s, 5

A dynamic function can also express task dependencies with native inputs:

@dynamic
def my_dynamic_subwf(a: int, b: int) -> int:
    x = t1(a=a)
    return t2(b=b, x=x)

PythonFunctionTask.execute routes dynamic tasks to dynamic_execute. In local execution, dynamic_execute creates or reuses a PythonFunctionWorkflow and executes it locally. In TASK_EXECUTION mode, it calls compile_into_workflow and returns a DynamicJobSpec for Flyte to run. In LOCAL_TASK_EXECUTION mode it directly invokes the function. Missing or unsupported execution state raises ValueError.

Pass node_dependency_hints only for dynamic tasks. Supplying it to a static task raises ValueError, because static workflow dependencies are discovered automatically. Dynamic compilation currently rejects reference tasks inside the generated workflow. The source documentation also cautions that a loop can produce very large generated workflows and recommends keeping dynamic workflows under fifty tasks.

Asynchronous tasks​

For a coroutine function, the decorator selects AsyncPythonFunctionTask. Its __call__ uses async_flyte_entity_call_handler, and async_execute awaits the user function in default execution mode. Async dynamic execution is deliberately unsupported: the dynamic branch raises NotImplementedError.

Eager workflows​

Use @eager when Python async code should call Flyte entities eagerly rather than compile the function into a workflow specification:

from flytekit import task, eager


@task
def add_one(x: int) -> int:
    return x + 1


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


@eager
async def eager_workflow(x: int) -> int:
    out = add_one(x=x)
    return double(x=out)


if __name__ == "__main__":
    import asyncio

    result = asyncio.run(eager_workflow(x=1))
    print(f"Result: {result}")  # "Result: 4"

EagerAsyncPythonFunctionTask removes any caller-provided execution mode and forces PythonFunctionTask.ExecutionBehavior.EAGER. It also sets TaskMetadata.is_eager=True. Local execution switches to EAGER_LOCAL_EXECUTION; backend execution uses EAGER_EXECUTION and a Controller worker queue. If no worker queue exists, the task constructs a remote from the execution context, installs SIGINT and SIGTERM handlers, and uses the current execution ID for its tag. _F_EE_ROOT is read to preserve the root eager execution tag for nested eager work.

The eager source example notes that FlyteRemote client credentials may be required for client-credentials authentication unless using a sandbox demo cluster. Backend eager execution requires an execution ID in the context. Eager and dynamic modes do not mix.

For failures, get_as_workflow adds an EagerFailureHandlerTask. Its fixed EagerFailureTaskResolver serializes as ['eager', 'failure', 'handler']. The handler is remote-only: it obtains the current project, domain, and execution name, finds active child executions tagged eager-exec, terminates them, and polls until none remain. It returns the input literal map unchanged and its execute method raises because it is expected to run through remote dispatch_execute.

Edge cases when extending or composing tasks​

  • ignore_input_vars removes names from the generated PythonFunctionTask interface. Those values are not ordinary Flyte inputs and are intended for client-side injection.
  • A no-output task returns VoidPromise; output count must otherwise match the declared interface.
  • Output conversion rejects a tuple as an individual output, while preserving the special handling for a one-element NamedTuple interface.
  • PythonTask rejects both deck flags together, warns that disable_deck is deprecated, defaults decks off, and empties deck_fields when decks are disabled.
  • Map and array-node wrappers accept only supported PythonFunctionTask/PythonInstanceTask forms and currently limit the wrapped task to zero or one output; dynamic and eager Python function tasks are rejected by ArrayNodeMapTask.
  • kwtypes is not an execution mechanism. It creates an insertion-ordered OrderedDict[str, Type] used for structured schema annotations and reference-entity interfaces, for example kwtypes(a=str, b=int).

The extracted core source tree contains the task, dynamic, eager, map, resolver, and task-mocking examples in module docstrings and embedded source examples rather than separate test/example files. Those examples are the authoritative usage patterns for the APIs shown here.