Skip to main content

Task authoring and execution

A Flyte task has two lives: you call it with Python values while authoring or testing a workflow, and Flyte later rehydrates it in a task container and invokes it with Flyte literals. flytekit separates those concerns through a task hierarchy and a dispatch path that converts between the two representations.

The task hierarchy​

The layers add progressively more Python- and execution-specific behavior:

Task
└── PythonTask
    └── PythonAutoContainerTask
        ├── PythonFunctionTask
        │   ├── AsyncPythonFunctionTask
        │   └── EagerAsyncPythonFunctionTask
        └── PythonInstanceTask

Task is the IDL-level abstraction. Its constructor stores a task type, name, typed interface, TaskMetadata, task-type version, security context, and documentation, and registers the instance in FlyteEntities.entities. It deliberately exposes no Python-native interface: python_interface returns None. Subclasses implement dispatch_execute, pre_execute, and execute.

PythonTask adds flytekit's Interface, task configuration, environment variables, and Deck settings. It also supplies the normal Python execution pipeline. PythonAutoContainerTask (defined in flytekit/core/python_auto_container.py) adds the container and default task-resolution behavior used by ordinary Python tasks. PythonFunctionTask is the function-backed implementation used by @task.

For tasks that have a Python interface but no Python function body, use a PythonInstanceTask subclass. Its class documentation shows the intended shape:

x = MyInstanceTask(name="x", .....)
x(a=5)

Such a class overrides execute; platform/plugin code supplies the behavior. PythonInstanceTask is used as the base for extensions such as ShellTask, DBTRun, DBTTest, DuckDBQuery, NotebookTask, and GreatExpectationsTask.

Declare a typed Python task​

Use the task decorator for the usual function-backed case. A function's annotations become its Flyte interface, so the smallest useful declaration is:

from flytekit import task

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

The decorator is implemented in flytekit/core/task.py. It accepts task configuration such as task_config, cache, retries, timeout, interruptible, container_image, environment, resource settings, execution_mode, task_resolver, enable_deck, pod_template, and pod_template_name. With no plugin configuration, it creates a PythonFunctionTask; with a registered configuration type, it selects the corresponding plugin class.

PythonFunctionTask.__init__ calls transform_function_to_interface(task_function, Docstring(callable_=task_function), pickle_untyped=pickle_untyped). It then removes any names in ignore_input_vars, extracts the function's module/name, and passes the resulting interface to PythonAutoContainerTask. A named multi-output annotation gives outputs stable names:

import typing
from flytekit import task

@task
def my_task(a: int) -> typing.NamedTuple("OutputsBC", b=int, c=str):
    return a + 2, "hello world"

assert my_task(a=3) == (5, "hello world")

The test tests/flytekit/unit/core/test_type_hints.py uses this declaration and verifies the local result (5, "hello world"). A function with no return annotation/output produces no outputs; local calls return None, while compilation uses a VoidPromise.

Calling a task does not simply call the function. Task.__call__ delegates to flyte_entity_call_handler, which chooses the promise-aware behavior for the current Flyte context. In a workflow compilation context, PythonTask.compile calls create_and_link_node to create the node. In local execution, Task.local_execute converts native values or Promises into a LiteralMap, executes the task, and wraps the resulting literals back into Promise objects (or a VoidPromise for a task with no outputs).

Configure execution with TaskMetadata​

Pass execution policy through @task arguments, or construct TaskMetadata when creating a class-based task:

from flytekit import task

@task(
    retries=2,
    timeout=60,
    interruptible=True,
    cache=True,
    cache_version="v1",
    cache_serialize=True,
)
def process(value: str) -> str:
    return value.strip()

TaskMetadata is defined in flytekit/core/base_task.py. Its fields include retries, timeout, interruptibility, deprecation text, cache settings, pod_template_name, generates_deck, and is_eager. An integer timeout is converted to datetime.timedelta(seconds=timeout) in __post_init__. to_taskmetadata_model() maps these settings to the Flyte task model, including the SDK runtime metadata and retry strategy.

The cache-related fields are validated immediately:

  • cache=True requires a non-empty cache_version.
  • cache_serialize=True requires cache=True.
  • cache_ignore_input_vars requires cache=True.

For example, @task(cache=True) raises ValueError during task construction. During local execution, Task.local_execute uses LocalConfig.auto() and LocalTaskCache: it looks up a cached literal map by task name, cache version, inputs, and ignored input names, and stores the result after a cache miss. Local caching is enabled by default and can be disabled with FLYTE_LOCAL_CACHE_ENABLED=false; FLYTE_LOCAL_CACHE_OVERWRITE forces execution instead of accepting a cache hit.

The preferred decorator form uses the Cache object from flytekit.core.cache:

from flytekit import Cache, task

@task(cache=Cache(version="a-version", serialize=True))
def process(value: str) -> str:
    return value.strip()

The cache tests also use Cache(version="a-version", ignored_inputs=("a",)) and policy-based Cache objects. Do not combine a Cache object with the legacy cache_version, cache_serialize, or cache_ignore_input_vars arguments; the decorator rejects those conflicting settings. The older cache=True plus cache_version=... form remains supported.

Deck configuration is handled by PythonTask. Set enable_deck=True and optionally choose deck_fields; the default fields are source code, dependencies, timeline, input, and output. disable_deck is deprecated (the constructor emits a FutureWarning), and setting both disable_deck and enable_deck raises ValueError. pod_template_name is metadata naming an existing cluster PodTemplate; it is separate from the pod_template argument, which configures the pod during task construction.

What happens during execution​

For a PythonTask, dispatch_execute in flytekit/core/base_task.py is the boundary between Flyte's literal types and user Python:

LiteralMap
  -> pre_execute(user_params)
  -> TypeEngine.literal_map_to_kwargs(...)
  -> execute(**native_inputs)
  -> post_execute(user_params, native_outputs)
  -> TypeEngine.async_to_literal(...)
  -> LiteralMap

pre_execute runs before input conversion, allowing an extension to prepare user-space state. The base implementation returns the parameters unchanged. dispatch_execute then calls _literal_map_to_python_input, which uses TypeEngine and the task's Python input types. It invokes execute, runs post_execute, and converts outputs through _output_to_literal_map. Output conversion handles zero outputs, one output, and multiple outputs, including the single-element NamedTuple convention. Conversion failures are wrapped as non-recoverable system errors for remote execution; local execution preserves the local exception with task context. Exceptions raised by user code become FlyteUserRuntimeException remotely and are re-raised with task context locally.

The runtime entry point in flytekit/bin/entrypoint.py loads the task, downloads the input LiteralMap, and calls task_def.dispatch_execute. It writes literal outputs, futures for a DynamicJobSpec, or an error document. To intentionally suppress output writing, raise IgnoreOutputs from flytekit.core.base_task rather than returning it:

from flytekit import task
from flytekit.core.base_task import IgnoreOutputs

@task
def worker() -> None:
    # Do work whose outputs are owned by another participant.
    raise IgnoreOutputs()

The entry point catches the resulting FlyteUserRuntimeException, checks whether its value is IgnoreOutputs, logs the condition, and returns without uploading outputs.pb. The exception's docstring identifies distributed training and peer-to-peer algorithms as the use case.

Dynamic function tasks​

Use @dynamic when the function body creates task nodes based on runtime values:

import typing
from flytekit import dynamic, task

@task
def t1(a: int) -> str:
    a = a + 2
    return "fast-" + str(a)

@dynamic
def ranged_int_to_str(a: int) -> typing.List[str]:
    s = []
    for i in range(a):
        s.append(t1(a=i))
    return s

assert ranged_int_to_str(a=5) == ["fast-2", "fast-3", "fast-4", "fast-5", "fast-6"]

flytekit/core/dynamic_workflow_task.py defines dynamic as a partial application of task with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC. In that mode, PythonFunctionTask.execute calls dynamic_execute instead of the function directly. During a real task execution, dynamic_execute calls compile_into_workflow, which builds a PythonFunctionWorkflow, serializes it, collects its task templates and nodes, and returns a DynamicJobSpec. The entry point writes that spec as futures. During local execution, flytekit executes the generated workflow and converts its native results into a literal map.

Set node_dependency_hints only for dynamic tasks when Flyte cannot infer dependencies before runtime. Supplying it to a default/static task raises ValueError. compile_into_workflow also rejects reference tasks inside dynamic tasks. Dynamic and eager behavior are different execution models: eager tasks do not use this runtime workflow compilation path.

Async and eager tasks​

The task decorator selects an async-capable task for an async function. AsyncPythonFunctionTask overrides __call__ with async_flyte_entity_call_handler; its async_execute awaits the wrapped function, and execute is backed by loop_manager.synced. It does not support dynamic execution: async_execute raises NotImplementedError for ExecutionBehavior.DYNAMIC.

For an eager workflow, decorate an async function with @eager:

from flytekit import eager, task

@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}")

The example is from the eager docstring in flytekit/core/task.py. The decorator constructs an EagerAsyncPythonFunctionTask, always enables Decks, and forces ExecutionBehavior.EAGER; its metadata has is_eager=True. Locally, async_execute switches to EAGER_LOCAL_EXECUTION and calls the function. For a backend execution, the task creates a Controller backed by a FlyteRemote; each Flyte entity call becomes a child execution whose result is awaited. Eager execution therefore requires the remote/authentication configuration described by the eager decorator; a sandbox cluster does not require the client-secret arguments.

get_as_workflow() wraps an eager task in an ImperativeWorkflow and adds an EagerFailureHandlerTask as its failure handler. That handler uses the Admin client to find unfinished child executions tagged with the parent eager execution and terminates them.

Extend function tasks with plugins​

Plugin authors can preserve the normal Python conversion pipeline while adding backend-specific setup and serialization. The Ray plugin in plugins/flytekit-ray/flytekitplugins/ray/task.py is a complete example:

@dataclass
class RayJobConfig:
    worker_node_config: typing.List[WorkerNodeConfig]
    head_node_config: typing.Optional[HeadNodeConfig] = None
    enable_autoscaling: bool = False
    runtime_env: typing.Optional[dict] = None
    address: typing.Optional[str] = None


class RayFunctionTask(PythonFunctionTask):
    _RAY_TASK_TYPE = "ray"

    def __init__(self, task_config: RayJobConfig, task_function: Callable, **kwargs):
        super().__init__(
            task_config=task_config,
            task_type=self._RAY_TASK_TYPE,
            task_function=task_function,
            **kwargs,
        )

    def pre_execute(self, user_params: ExecutionParameters) -> ExecutionParameters:
        ray.init(address=self._task_config.address)
        return user_params

TaskPlugins.register_pythontask_plugin(RayJobConfig, RayFunctionTask)

The actual Ray implementation also overrides get_custom to serialize the Ray cluster specification. TaskPlugins.register_pythontask_plugin maps the configuration type to the task class; consequently, @task(task_config=RayJobConfig(...)) can select RayFunctionTask instead of the base PythonFunctionTask. Only one different plugin may be registered for a given configuration type.

Use PythonInstanceTask rather than PythonFunctionTask when the extension has no user function body. It still participates in the Python task interface and container machinery, but the subclass supplies the platform-defined execute implementation.

Rehydrate tasks with a resolver​

Serialization and execution are separated by TaskResolverMixin. At serialization time, a resolver supplies a location and loader_args; at runtime, pyflyte-execute imports the resolver and calls load_task to reconstruct the task. The TaskResolverMixin docstring shows the default command shape:

pyflyte-execute --inputs s3://path/inputs.pb --output-prefix s3://outputs/location \
  --raw-output-data-prefix /tmp/data \
  --resolver flytekit.core.python_auto_container.default_task_resolver \
  -- \
  task-module repo_root.workflows.example task-name t1

The default resolver imports repo_root.workflows.example and looks up t1. This is why a normal function task must be accessible at module level when the default resolver is used. PythonFunctionTask.__init__ rejects nested, inner, or local functions with ValueError (test modules beginning with test_ are allowed); use functools.wraps/update_wrapper for a decorating wrapper or provide a custom TaskResolverMixin.

A custom resolver implements location, name, load_task(loader_args), loader_args(settings, task), and get_all_tasks. It may also override task_name. Class-based resolver implementations and special resolvers such as EagerFailureTaskResolver use this contract.

Interfaces for non-function tasks​

Function tasks derive interfaces from annotations. For a task with no function signature, use kwtypes to build an ordered mapping of names to Python types and pass it to the task constructor. The SQL task test uses the following real pattern:

import datetime
from flytekit import FlyteSchema, SQLTask, TaskMetadata, kwtypes

sql = SQLTask(
    "my-query",
    query_template="SELECT * FROM hive.city WHERE ds = '{{ .Inputs.ds }}' LIMIT 10",
    inputs=kwtypes(ds=datetime.datetime),
    outputs=kwtypes(results=FlyteSchema),
    metadata=TaskMetadata(retries=2, cache=True, cache_version="0.1"),
)

kwtypes(ds=datetime.datetime) produces an OrderedDict mapping ds to datetime.datetime. This is the explicit-interface counterpart to transform_function_to_interface for @task functions.