Skip to main content

Task authoring and execution

Task authoring model​

A Flyte task is a callable entity with a typed interface, metadata, and a runtime representation. For the usual Python case, decorate a module-level function with @task:

from flytekit import task

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

The decorator creates a PythonFunctionTask for the function. PythonFunctionTask.__init__ calls transform_function_to_interface to derive the Flyte interface from the function annotations, and extract_task_module to derive the task name from the function's module and name. The resulting object, rather than the original function, is what you call from a workflow or another Flyte entity.

Task is the lowest-level abstraction. It stores the task type, name, IDL TypedInterface, TaskMetadata, optional security context, and documentation. Construction also appends the task to FlyteEntities.entities, so module-level task definitions are registered when their module is imported. Task.__call__ delegates invocation to flyte_entity_call_handler; the handler can therefore produce workflow promises during compilation or execute the task locally, depending on the current FlyteContext.

PythonTask adds the Python-native Interface layer on top of Task. It exposes Python input and output types, optional plugin configuration through task_config, an environment dictionary, and deck settings. PythonInstanceTask is the alternative for class-based tasks whose execution logic is supplied by the class rather than by a user function:

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

PythonInstanceTask is intended for platform-defined execute methods. PythonFunctionTask is the appropriate base when a task wraps a user-defined Python function.

Configure task metadata and runtime settings​

Pass task-level policies through @task. For example, tests/flytekit/unit/core/test_python_function_task.py declares a cache with serialization enabled:

@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 holds the values that become the task metadata model: caching fields, retries, timeout, interruptibility, deprecation text, pod-template name, deck generation, and eager-task state. TaskMetadata.to_taskmetadata_model() converts these values to Flyte's task model, including the SDK runtime metadata and retry strategy.

The metadata validation is strict. cache=True requires a non-empty cache_version; cache_serialize=True requires caching to be enabled; and cache_ignore_input_vars also requires caching. Each invalid combination raises ValueError during task construction. An integer timeout is interpreted as seconds and converted to datetime.timedelta; another value must already be a datetime.timedelta.

Flytekit also supports the Cache configuration object for the newer cache API:

@task(cache=Cache(policies=SaltCachePolicy()))
def t1(a: int) -> int:
    return a

@task(cache=Cache(version="a-version", serialize=True))
def t3(a: int) -> int:
    return a

@task(cache=Cache(version="a-version", ignored_inputs=("a",)))
def t4(a: int) -> int:
    return a

The task decorator also passes execution settings used by PythonAutoContainerTask, including container_image, environment, resources such as limits=Resources(mem="200Mi"), secrets, and pod-template settings. For a pod-template task, the test suite verifies that get_container() returns None and get_k8s_pod() carries the Kubernetes pod definition:

@task(
    container_image="repo/image:0.0.0",
    pod_template=PodTemplate(
        primary_container_name="primary",
        labels={"lKeyA": "lValA"},
        annotations={"aKeyA": "aValA"},
        pod_spec=V1PodSpec(
            containers=[V1Container(name="primary")]
        ),
    ),
    pod_template_name="A",
)
def func_with_pod_template(i: str):
    print(i + "a")

PythonTask.construct_node_metadata() carries timeout, retry, and interruptible values into a workflow node. PythonTask.compile() then calls create_and_link_node, which is how a task invocation becomes a node when it is used in a workflow definition.

Deck generation is configured with enable_deck and deck_fields. PythonTask defaults its deck field tuple to source code, dependencies, timeline, input, and output, but the constructor leaves the effective field list empty unless deck output is enabled. Supplying both disable_deck and enable_deck raises ValueError; disable_deck is retained as a deprecated spelling, so use enable_deck for new task definitions.

What execution does with inputs and outputs​

A local call follows the same task dispatch boundary used by runtime execution:

Task.__call__
  -> flyte_entity_call_handler
  -> Task.local_execute (local calls)
  -> PythonTask.dispatch_execute
       -> pre_execute
       -> LiteralMap -> Python native inputs
       -> execute(**native_inputs)
       -> post_execute
       -> Python native outputs -> LiteralMap

Task.local_execute() first uses translate_inputs_to_literals to normalize native arguments, promises, lists, and dictionaries into a Flyte LiteralMap. If local caching is enabled and the task metadata requests caching, it looks up the task name, cache version, literal inputs, and ignored input variables in LocalTaskCache. A cache hit skips sandbox_execute; a miss executes the task and stores the resulting literal map.

PythonTask.dispatch_execute() calls pre_execute() before input conversion. This hook is where task implementations can modify execution parameters before type transformers run. It then uses _literal_map_to_python_input() and TypeEngine.literal_map_to_kwargs() to create native keyword arguments, invokes execute(**native_inputs), and calls post_execute(). Finally, _output_to_literal_map() maps returned values to declared output names and asynchronously converts each value with TypeEngine.async_to_literal().

The output-shape rules are part of PythonTask: zero declared outputs produce an empty map, one output accepts the returned value directly, and multiple outputs are mapped by position. A one-element NamedTuple has a special case so its tuple value is assigned to the single declared output. Conversion failures include the task name and output position in the raised error.

For a normal PythonFunctionTask, execute() is deliberately small:

def execute(self, **kwargs) -> Any:
    if self.execution_mode == self.ExecutionBehavior.DEFAULT:
        return self._task_function(**kwargs)
    elif self.execution_mode == self.ExecutionBehavior.DYNAMIC:
        return self.dynamic_execute(self._task_function, **kwargs)

The function itself receives native Python values. The surrounding PythonTask machinery is responsible for translating Flyte literals before the call and translating the return value afterward. pre_execute and post_execute remain extension points: the base PythonTask.pre_execute returns the supplied execution parameters unchanged, while post_execute returns the result unchanged.

A task that has no meaningful output can return normally with no outputs. For distributed cases where only one worker should publish results, raise IgnoreOutputs rather than returning it. IgnoreOutputs is an exception sentinel handled by the task entrypoint; raising it causes output writing to be skipped. It must propagate out of the task execution path—catching it inside the task body prevents the entrypoint from observing the signal.

Select normal, dynamic, or eager execution​

PythonFunctionTask.ExecutionBehavior has three values: DEFAULT, DYNAMIC, and EAGER. The decorators select the specialized behavior for you.

Normal tasks​

@task produces the default mode. The function executes as the task body when dispatch_execute() runs. Local workflow calls still return Flyte promises while the workflow is being compiled; direct local execution eventually unwraps inputs, calls the function, and wraps outputs back into promises or a VoidPromise.

The local cache behavior is observable in tests/flytekit/unit/core/test_local_cache.py:

@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

@workflow
def check_evenness(n: int) -> bool:
    return is_even(n=n)

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

The second local call returns the cached literal result, so the function body is not called a second time when local caching is enabled.

Dynamic tasks​

Use @dynamic when the function body constructs a workflow at execution time. The body can create a variable number of task invocations based on runtime values:

@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"]

@dynamic creates a PythonFunctionTask with ExecutionBehavior.DYNAMIC. In a real task execution, dynamic_execute() calls compile_into_workflow(). That method builds or reuses a PythonFunctionWorkflow, serializes its generated nodes and task templates, and returns a DynamicJobSpec. In local execution, dynamic_execute() runs the generated workflow locally and converts its outputs to a literal map. Dynamic functions and normal functions therefore share the same task interface, but their runtime result is either native local output or a dynamically compiled job specification.

node_dependency_hints is only valid with dynamic execution. Supplying it to a regular task raises ValueError, because static tasks and workflows allow Flytekit to determine dependencies during ordinary compilation. Keep dynamic expansion bounded: the generated DynamicJobSpec contains the produced nodes and task templates, so a very large loop creates a correspondingly large specification.

Eager tasks​

Use @eager on an asynchronous function when Flyte entities inside the function should be executed as the function runs rather than compiled into one workflow specification:

@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 sets TaskMetadata.is_eager=True and enables decks by default. Its run_with_backend() method runs the function with an eager execution state and a Controller; each Flyte entity call is represented by a backend execution, and the controller renders the eager execution deck. get_as_workflow() can wrap the eager task in an ImperativeWorkflow and attaches an EagerFailureHandlerTask failure node. That cleanup task uses EagerFailureTaskResolver and queries Flyte Admin to terminate still-running eager sub-executions.

Eager and dynamic behavior are not interchangeable. AsyncPythonFunctionTask.async_execute() awaits a default-mode function, but raises NotImplementedError for dynamic mode; EagerAsyncPythonFunctionTask is the eager-specific implementation.

Async Python functions​

Adding async to a task function changes the task class selected by the task decorator:

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

The decorator detects coroutine functions and uses AsyncPythonFunctionTask. Its __call__ is awaitable, and async_execute() awaits the wrapped function. The synchronous execute entry point is provided by loop_manager.synced, allowing the normal dispatch machinery to invoke an async task when it is running in a synchronous execution path.

Async and eager entities can be composed with ordinary Python async constructs. The async semantics test uses both a regular async task and concurrent eager calls:

@eager
async def base_wf(x: int) -> int:
    out = add_one(x=x)
    doubled = a_double(x=x)
    if out - await doubled < 0:
        return -1
    final = double(x=out)
    return final

@eager
async def parent_wf(a: int, b: int) -> typing.Tuple[int, int]:
    t1 = asyncio.create_task(base_wf(x=a))
    t2 = asyncio.create_task(base_wf(x=b))
    i1, i2 = await asyncio.gather(t1, t2)
    return i1, i2

Task resolution in the execution container​

Serialization and execution are separate phases. TaskResolverMixin defines the contract that connects them: a resolver supplies a location, a resolver name, loader_args(), load_task(), and get_all_tasks(). The resolver location is placed in the pyflyte-execute command, and the loader arguments identify the task to rehydrate in the worker.

The default resolver uses the defining module and task name. The pod-template test verifies the concrete command generated for func_with_pod_template:

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_function_task
 task-name func_with_pod_template

The line break before task-name above is only formatting; the actual argument list contains --, task-module, the module name, task-name, and the task variable name. Consequently, a normal task function must be importable at module scope when the default resolver is used. PythonFunctionTask rejects nested or local functions with ValueError, except for permitted test functions and wrapped module-level functions recognized through functools.wraps or functools.update_wrapper. If a task cannot use module-and-name lookup, implement a custom TaskResolverMixin and pass it as task_resolver.

Extend Python function tasks with plugins​

Plugin task types retain the Python function interface while adding backend-specific configuration and serialization. Register a configuration type with a PythonFunctionTask subclass through TaskPlugins.register_pythontask_plugin. For example, the Spark plugin defines PysparkFunctionTask as a PythonFunctionTask[Spark] and registers it for Spark, Databricks, and DatabricksV2:

class PysparkFunctionTask(AsyncConnectorExecutorMixin, PythonFunctionTask[Spark]):
    _SPARK_TASK_TYPE = "spark"

TaskPlugins.register_pythontask_plugin(Spark, PysparkFunctionTask)
TaskPlugins.register_pythontask_plugin(Databricks, PysparkFunctionTask)
TaskPlugins.register_pythontask_plugin(DatabricksV2, PysparkFunctionTask)

When @task receives a registered task_config, TaskPlugins.find_pythontask_plugin() selects the matching class; an unregistered configuration falls back to PythonFunctionTask. A plugin can override the task-specific execution and serialization hooks supplied by the base abstractions, including pre_execute, execute, get_custom, get_container, or get_k8s_pod. For async plugin functions, subclass AsyncPythonFunctionTask so the plugin preserves the awaitable execution path.

Constraints to check before authoring a task​

  • Define default-resolver task functions at module scope. Nested and local functions cannot be rehydrated by module-and-name lookup.
  • Give cached tasks a cache version, and use the Cache object when configuring serialization or ignored inputs. Invalid metadata combinations fail during construction.
  • Raise IgnoreOutputs; do not return or swallow the sentinel when a distributed worker must omit output publication.
  • Use enable_deck, not the deprecated disable_deck, and do not provide both parameters.
  • Do not provide node_dependency_hints to a static task; it is restricted to dynamic tasks.
  • Remember that overriding PythonFunctionTask.execute() means taking responsibility for the execution behavior you override, including dynamic handling if the subclass is intended to remain a dynamic task.
  • Keep dynamic expansion bounded because runtime compilation produces a DynamicJobSpec containing the generated nodes and task templates.
  • Eager task execution installs signal handlers while constructing its controller; EagerAsyncPythonFunctionTask.execute() therefore performs that setup from the main execution thread.