Skip to main content

Task authoring and execution

A Flyte task has two related representations: the task object that participates in a workflow graph, and the Python code or plugin implementation that runs when Flyte executes it. Task provides the IDL-oriented identity and call boundary; PythonTask adds a typed native Python interface and the input/output conversion lifecycle; PythonFunctionTask and PythonInstanceTask provide the two common Python authoring styles. Function and instance tasks also inherit container and task-resolver behavior through PythonAutoContainerTask.

Declare a function task​

Use the public task decorator when the task has a Python function body. Type annotations supply the task interface, and keyword calls compose tasks in a workflow:

from flytekit import task, workflow

@task(cache=True, cache_version="1", retries=3)
def sum(x: int, y: int) -> int:
    return x + y

@task(cache=True, cache_version="1", retries=3)
def square(z: int) -> int:
    return z*z

@workflow
def my_workflow(x: int, y: int) -> int:
    return sum(x=square(z=x), y=square(z=y))

Inside a workflow, square(z=x) and sum(...) are task calls used to build the graph; they are not the mechanism by which the Python function bodies run during workflow declaration. The call handler used by Task.__call__ links the task into the current compilation context and represents its outputs as promises. A task can therefore be composed using ordinary-looking Python assignments and unpacking:

from flytekit import task, workflow

@task(enable_deck=True)
def t1(a: int) -> typing.NamedTuple("OutputsBC", t1_int_output=int, c=str):
    return a + 2, "world"

@task
def t2(a: str, b: str) -> str:
    return b + a

@workflow
def my_wf(a: int, b: str) -> (int, str):
    x, y = t1(a=a)
    d = t2(a=y, b=b)
    return x, d

The decorator implementation in flytekit/core/task.py constructs TaskMetadata, chooses a registered Python-task plugin from task_config, selects AsyncPythonFunctionTask for coroutine functions when the ordinary plugin is selected, instantiates the task, and applies update_wrapper to retain function metadata. PythonFunctionTask.__init__ then calls transform_function_to_interface on the callable and its docstring, removes any requested ignored inputs, and derives the task name from the function's module and name.

The function must be non-None and its annotations are used to discover the interface. ignore_input_vars removes named inputs from that interface; the constructor documents this for values injected on the client side. pickle_untyped=True is available for untyped outputs, but PythonFunctionTask explicitly describes that option as unsuitable for production use.

Configure execution metadata​

Pass execution policy through the decorator or through TaskMetadata when constructing a task directly. The README example uses caching and retries; other supported task options include timeout, interruptible, deprecated, environment, resources, secrets, container/image settings, pod-template settings, and deck configuration.

TaskMetadata stores the values that become Flyte task metadata. Its validation is immediate:

  • cache=True requires a non-empty cache_version when using the legacy metadata fields.
  • cache_serialize=True and cache_ignore_input_vars both require caching to be enabled.
  • An integer timeout is converted to datetime.timedelta seconds; a truthy value of another type raises ValueError.
  • retry_strategy exposes the configured retry count as Flyte's retry model.

For example, these legacy cache options are valid, while the missing version and missing cache cases are rejected:

from flytekit import task

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

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

The decorator also accepts the newer Cache object. When cache=True is used without a legacy version, task creates a Cache; when a Cache instance is supplied, it computes the version from the function and relevant container settings. Do not combine a Cache instance with cache_version, cache_serialize, or cache_ignore_input_vars; flytekit/core/task.py rejects that combination.

Deck configuration has two mutually exclusive spellings. Set enable_deck or the deprecated disable_deck, but not both. PythonTask defaults decks off; when enabled, its default fields include source code, dependencies, timeline, input, and output. PythonFunctionTask adds source-code and dependency rendering, while the shared Python-task lifecycle writes input and output decks.

Task-specific environment values are merged with SerializationSettings.env when the container is serialized, with the task values taking precedence. The serialized container also needs an image: PythonAutoContainerTask resolves an explicit image or ImageSpec, or uses SerializationSettings.image_config; no usable image is implicit when neither resolves.

Understand the task abstractions​

Task stores the task type, name, typed Flyte interface, metadata, security context, and documentation, and registers itself in FlyteEntities.entities. Its default python_interface is None, and its default get_input_types() cannot infer Python-native types. It is consequently the low-level abstraction for IDL-oriented tasks rather than the usual user-facing function task. Its serialization hooks for container, pod, SQL, custom data, config, and extended resources also return None unless a subclass supplies them.

Task.local_execute() handles local calls. It translates native values and promises into a Flyte LiteralMap, checks LocalTaskCache when task caching and local caching are enabled, invokes sandbox_execute() and therefore dispatch_execute(), then wraps the returned literals as Promise objects. A task with no declared outputs returns VoidPromise.

PythonTask supplies the missing Python layer. It transforms an Interface into a typed Flyte interface, keeps Python input/output types for TypeEngine, and creates workflow nodes in compile() through create_and_link_node. Its dispatch_execute() lifecycle is:

pre_execute(user parameters)
    -> LiteralMap to native Python inputs
    -> execute(**native_inputs)
    -> post_execute(...)
    -> native outputs to LiteralMap
    -> optional deck writing

Conversion errors are decorated with the task name. Output conversion handles zero outputs, a single output, multiple outputs, and the special single-element NamedTuple convention. An individual output that is itself a tuple is rejected, and the number of returned values must match the declared output interface.

Choose a function task or an instance task​

Use PythonFunctionTask when a callable is the task implementation. This is also the extension point for plugin-backed function tasks. For example, the Ray plugin subclasses it, supplies a custom task type and RayJobConfig, and uses pre_execute() to call ray.init():

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,
        )
        self._task_config = task_config

    def pre_execute(self, user_params: ExecutionParameters) -> ExecutionParameters:
        init_params = {"address": self._task_config.address}
        ray.init(**init_params)
        return user_params

The complete Ray implementation also adds plugin-specific serialized data through get_custom(). After registration with TaskPlugins.register_pythontask_plugin, a @task call with the corresponding task_config selects the plugin rather than the plain PythonFunctionTask.

Use PythonInstanceTask when there is no user-defined function body and a platform or plugin implementation supplies execute(). The dbt plugin declares an explicit interface with kwtypes, then implements the execution method:

class DBTRun(PythonInstanceTask):
    def __init__(self, name: str, **kwargs):
        super(DBTRun, self).__init__(
            task_type="dbt-run",
            name=name,
            task_config=None,
            interface=Interface(inputs=kwtypes(input=DBTRunInput), outputs=kwtypes(output=DBTRunOutput)),
            **kwargs,
        )

    def execute(self, **kwargs) -> DBTRunOutput:
        task_input: DBTRunInput = kwargs["input"]
        ...

kwtypes(input=DBTRunInput, output=DBTRunOutput) is an ordered mapping from names to Python types. It describes the interface; it does not execute a function. DBTRun is invoked like a normal task after construction, and its execute() implementation runs the dbt CLI and returns DBTRunOutput.

Hosted execution and task rehydration​

A hosted Python task must tell pyflyte-execute how to reconstruct its task object inside the worker. TaskResolverMixin defines that contract with location, name, load_task(loader_args), loader_args(settings, task), and get_all_tasks(). A custom resolver can implement these methods when the default module-and-variable lookup is not appropriate.

PythonAutoContainerTask.get_default_command() places the resolver location and loader arguments after the standard input/output arguments:

pyflyte-execute --inputs {{.input}} --output-prefix {{.outputPrefix}} \
  --raw-output-data-prefix {{.rawOutputDataPrefix}} \
  --checkpoint-path {{.checkpointOutputPrefix}} \
  --prev-checkpoint {{.prevCheckpointPrefix}} \
  --resolver <resolver.location> -- <resolver.loader_args>

The default resolver uses the module containing the task and the task variable name. Consequently, a PythonFunctionTask using that resolver must refer to an accessible module-level function. PythonFunctionTask rejects nested or local functions, except recognized test functions and properly wrapped module-level functions. If a custom decorator wraps the function, preserve its metadata with functools.wraps or functools.update_wrapper, or provide a custom TaskResolverMixin.

Select execution behavior​

PythonFunctionTask.ExecutionBehavior has DEFAULT, DYNAMIC, and EAGER modes. Normal @task uses DEFAULT: execute() calls the stored function directly. The ordinary dispatch path still performs context setup and literal/native conversion before and after that call.

Dynamic tasks​

Use @dynamic when the function body creates a workflow at runtime. The same function abstraction is used, but execute() delegates to dynamic_execute(). Locally, Flytekit creates or reuses a PythonFunctionWorkflow and executes it with native values; during task execution it compiles the generated workflow and returns a DynamicJobSpec (or a literal map when no nodes are produced).

@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

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

node_dependency_hints is valid only for dynamic tasks; the constructor rejects it for static/default tasks because their dependencies can be found during normal compilation. Dynamic compilation also rejects ReferenceTask entities inside the generated workflow. Dynamic and eager are separate modes: async function tasks explicitly raise NotImplementedError for dynamic async execution.

Async and eager tasks​

When the decorated function is a coroutine, task() selects AsyncPythonFunctionTask for the ordinary Python-function plugin. Its asynchronous call handler awaits the function in async_execute(), and execute is a synchronized wrapper around that method.

Use @eager for an async eager task. The repository's minimal eager composition uses a normal task as a child:

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

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

EagerAsyncPythonFunctionTask removes any supplied execution mode, forces ExecutionBehavior.EAGER, and sets TaskMetadata.is_eager=True; it also enables decks by default. Locally it switches to eager-local execution state and awaits the user function. For backend execution, it creates or uses a Controller worker queue, runs child Flyte entities as executions, renders an Eager Executions deck, and converts an EagerException into a non-recoverable system failure. Backend eager execution requires a user execution ID and a remote supplied by the Flyte plugin configuration. get_as_workflow() builds an imperative workflow and adds an EagerFailureHandlerTask cleanup node.

Constraints that commonly affect execution​

  • Use keyword-style task calls in workflow bodies. The resulting values are promises during compilation; local execution unwraps them and converts them through TypeEngine.
  • Map tasks accept only default-mode PythonFunctionTask or PythonInstanceTask and reject eager/dynamic function tasks. They also reject tasks with more than one output.
  • A function task with no outputs returns VoidPromise; returning None is not the same as raising IgnoreOutputs.
  • Raise IgnoreOutputs only for the exceptional distributed-training or peer-to-peer path described by its class docstring. The entrypoint catches that marker and deliberately skips writing outputs.pb.
  • Configure an image through container_image/ImageSpec or SerializationSettings.image_config before hosted serialization. Configure SerializationSettings.env and task environment deliberately because task-level values override collisions.
  • For output failures, inspect the declared interface first: PythonTask checks output count, treats a one-element NamedTuple specially, rejects tuple-valued individual outputs, and reports the task name and output position in conversion errors.