Skip to main content

Task authoring and execution

A Flyte task is both a callable authoring object and a serialized execution target. When you write @task, Flytekit builds a task object from the function’s annotations, metadata, and execution settings; when you call that object in a workflow, the call is compiled into a node rather than immediately running the function. During local execution, the same call path converts values to Flyte literals, dispatches the task, and wraps its outputs back into native values or Promise objects.

Start with an annotated Python function​

The usual entry point is the public task decorator. Annotations define the task’s Python interface, and calls use keyword arguments:

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

This is the example in README.md. task() creates a PythonFunctionTask for a synchronous function, and update_wrapper copies the decorated function’s metadata onto the task instance. The task object remains callable because Task.__call__ delegates to Flytekit’s entity call handler. In a workflow-compilation context, that handler creates and links a node; in local execution, it follows the task’s local execution path. Consequently, square(z=x) is a workflow value during compilation, while the function body is run with native values when the workflow is locally executed.

task() also selects the appropriate implementation. It creates a TaskMetadata object, looks up a registered Python-task plugin using the task configuration type, and instantiates the selected plugin with the function, interface-related options, container settings, resources, resolver, and metadata. Coroutine functions are routed to AsyncPythonFunctionTask when the selected plugin is the ordinary PythonFunctionTask.

PythonFunctionTask.__init__ derives the native interface with transform_function_to_interface, using both annotations and the function docstring. It removes any names supplied through ignore_input_vars, derives the task name and module from the callable, and passes the resulting interface to PythonAutoContainerTask. A function-backed task therefore has two related interfaces:

  • python_interface contains Python input and output types.
  • interface is the Flyte typed interface used for literal conversion and serialization.

Use a module-level function with the default resolver. PythonFunctionTask rejects None as task_function, and it rejects nested or local functions outside test modules because the default resolver must be able to import the module and find the task name. If a custom decorator wraps a function, use functools.wraps or functools.update_wrapper; alternatively provide a resolver implementing TaskResolverMixin.

Configure task behavior separately from function inputs​

Task settings are not ordinary function arguments. They become task metadata, container, pod, or plugin configuration. For example, the decorator accepts settings such as retries, timeout, interruptible, container_image, environment, resources, secret_requests, pod_template, pod_template_name, and deck options.

TaskMetadata in flytekit/core/base_task.py stores execution metadata:

from flytekit.core.base_task import TaskMetadata

metadata = TaskMetadata(
    cache=True,
    cache_serialize=True,
    cache_version="1.0",
    retries=2,
    timeout=300,
)

An integer timeout is interpreted as seconds and converted to datetime.timedelta. Other timeout values must already be a datetime.timedelta or TaskMetadata raises ValueError. Caching has explicit validation: cache=True requires a non-empty cache_version; cache_serialize=True and cache_ignore_input_vars are invalid unless caching is enabled. The decorator’s newer Cache form computes the version, but it cannot be combined with the deprecated cache_version, cache_serialize, or cache_ignore_input_vars arguments.

The README uses the legacy-compatible form directly on a task:

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

TaskMetadata.retry_strategy converts the retry count to Flyte’s retry model. to_taskmetadata_model() serializes the cache, retry, timeout, interruptibility, deprecation message, deck, pod-template, and eager fields and includes Flytekit’s SDK runtime metadata and version.

For hosted Python tasks, PythonAutoContainerTask resolves the image using SerializationSettings.image_config. Task-level environment is merged with SerializationSettings.env; task values take precedence. A pod template causes the task to serialize a Kubernetes pod rather than an ordinary container. Deck generation is disabled by default in PythonTask, despite the default deck_fields list. Pass enable_deck=True to activate it. disable_deck is deprecated, and supplying both disable_deck and enable_deck raises ValueError.

Local caching is a separate runtime choice. Task.local_execute() consults LocalTaskCache only when both TaskMetadata.cache and LocalConfig.cache_enabled are true. LocalConfig.cache_overwrite=True skips a cache hit, executes the task, and stores the new result. A cache lookup uses the task name, cache version, literalized inputs, and cache_ignore_input_vars.

Follow the execution pipeline​

The core abstraction is Task in flytekit/core/base_task.py. It stores the task type, name, Flyte typed interface, metadata, optional security context, and documentation, and registers each instance in FlyteEntities. Its abstract methods define the execution contract:

  • pre_execute(user_params) runs before input conversion.
  • execute(**kwargs) runs the task body.
  • dispatch_execute(ctx, input_literal_map) connects Flyte literals to execution.

Task.local_execute() accepts native values or Promise values. It calls translate_inputs_to_literals() using the task’s Flyte interface and native input types, then invokes sandbox_execute(). The sandbox creates a task-specific execution context and calls dispatch_execute(). After execution, local_execute() checks the declared output count, constructs Promise objects for outputs, and returns a VoidPromise when the task declares no outputs.

PythonTask supplies the Python-native layer used by function tasks and plugin tasks. Its dispatch_execute() performs the following data flow:

LiteralMap
  -> PythonTask._literal_map_to_python_input()
  -> TypeEngine.literal_map_to_kwargs()
  -> execute(**native_inputs)
  -> post_execute()
  -> PythonTask._output_to_literal_map()
  -> LiteralMap

Before converting inputs, it calls pre_execute. It invokes execute(**native_inputs), calls post_execute, and converts returned values asynchronously through TypeEngine.async_to_literal. A returned LiteralMap or DynamicJobSpec is passed through rather than converted again. Output conversion follows the declared output names; a single-output NamedTuple receives special handling, while a tuple returned for an individual output is rejected. Conversion and execution failures are reported differently by mode: local execution preserves the original exception with task context, while hosted execution wraps user failures in FlyteUserRuntimeException and other failures in FlyteNonRecoverableSystemException. Flyte upload and download transformer errors are re-raised unchanged.

PythonFunctionTask.execute() is the function-specific implementation:

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 default mode calls the user function directly. pre_execute is a no-op unless overridden, and post_execute returns the result unchanged unless a subclass overrides it. These hooks are the extension points for changing execution parameters before conversion or cleaning up and altering results afterward.

A task may raise IgnoreOutputs when its generated outputs can safely be ignored. The exception is handled by flytekit/bin/entrypoint.py, which skips writing outputs.pb; downstream consumers therefore cannot rely on outputs from such an execution. The exception is specifically documented for distributed-training and peer-to-peer algorithms.

Understand hosted serialization and rehydration​

A hosted worker must reconstruct the task object before it can call dispatch_execute. TaskResolverMixin defines that rehydration contract through location, loader_args(), load_task(), name(), and get_all_tasks(). PythonAutoContainerTask.get_default_command() uses those methods to construct the worker command:

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

For the default resolver, the loader arguments identify the importable module and task name. The pod-template test in tests/flytekit/unit/core/test_python_function_task.py verifies the concrete form:

[
    "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",
]

Use a custom TaskResolverMixin when module-and-name lookup is insufficient—for example, for a custom task representation or a task that cannot be loaded by the default module-level resolver. At runtime, the entrypoint loads the task from the resolver, calls dispatch_execute, and handles output writing or IgnoreOutputs.

Choose an extension layer​

Use Task when you need the IDL-facing base abstraction without a Python-native interface. Use PythonTask when the task has typed Python inputs and outputs but its execution is supplied by a plugin or subclass rather than a user function. PythonTask exposes task_config, type lookup, node metadata construction, compilation, literal conversion, and deck handling.

For a class-backed task with no user-defined function body, subclass PythonInstanceTask. Its constructor forwards the name, typed task configuration, task type, resolver, and other options to PythonAutoContainerTask; the subclass supplies execute. This is the appropriate shape for a platform-defined execution method.

For a function-backed plugin, subclass PythonFunctionTask with a typed configuration and register the pair with TaskPlugins. GenericSparkTask in plugins/flytekit-spark/flytekitplugins/spark/generic_task.py demonstrates the pattern:

@dataclass
class GenericSparkConf(object):
    main_class: str
    applications_path: str
    spark_conf: Optional[Dict[str, str]] = None
    hadoop_conf: Optional[Dict[str, str]] = None

class GenericSparkTask(PythonFunctionTask[GenericSparkConf]):
    def __init__(self, task_config: GenericSparkConf, task_function: Callable, ...):
        super(GenericSparkTask, self).__init__(
            task_config=task_config,
            task_type="spark",
            task_function=task_function,
            ...,
        )

    def get_custom(self, settings: SerializationSettings) -> Dict[str, Any]:
        job = SparkJob(...)
        return MessageToDict(job.to_flyte_idl())

TaskPlugins.register_pythontask_plugin(GenericSparkConf, GenericSparkTask)

The subclass keeps Python function discovery and execution from PythonFunctionTask, while get_custom() supplies plugin-specific serialized data. Other task types can override get_container, get_k8s_pod, get_sql, get_config, or get_extended_resources on the base task layers.

Use dynamic, async, and eager execution modes deliberately​

Dynamic tasks​

@dynamic is implemented as task(..., execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC). The function can use native control flow and create task calls at runtime:

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

In a local execution context, dynamic_execute() creates or reuses a PythonFunctionWorkflow, executes it with native inputs, and returns a LiteralMap. In TASK_EXECUTION mode, compile_into_workflow() compiles the generated workflow and returns a DynamicJobSpec containing its task templates, nodes, outputs, and subworkflows. node_dependency_hints may be supplied for dynamic tasks, but PythonFunctionTask rejects them for static tasks because static dependencies can be found automatically.

Keep the execution-mode restriction in mind when composing with other wrappers: array-node map tasks accept only PythonFunctionTask instances in DEFAULT mode, not dynamic or eager tasks.

Async tasks​

Decorate an async function with @task and await its result through the async call handler. AsyncPythonFunctionTask.__call__ is awaitable, and async_execute() awaits the underlying function. Async dynamic execution is explicitly not implemented in AsyncPythonFunctionTask; eager and dynamic execution modes are not combined there.

Eager workflows​

@eager creates an EagerAsyncPythonFunctionTask. Its constructor forces ExecutionBehavior.EAGER, sets TaskMetadata.is_eager=True, and defaults enable_deck=True.

@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

res = await simple_eager_workflow(x=1)

Locally, the eager task changes the execution state and awaits the function. During a real execution, execute() ensures a Controller worker queue exists; run_with_backend() runs the function with EAGER_EXECUTION state, awaits child task executions, and renders an Eager Executions deck. The remote path requires a user-space execution ID. Child execution tagging uses the current execution name and _F_EE_ROOT to propagate the root eager execution name.

get_as_workflow() wraps an eager task in an ImperativeWorkflow and adds an EagerFailureHandlerTask failure node. The cleanup task has a fixed EagerFailureTaskResolver with loader arguments ['eager', 'failure', 'handler']. Its remote dispatch_execute() lists executions tagged for the eager parent in UNDEFINED, QUEUED, or RUNNING phases, terminates them, and returns the input literal map; its execute() raises because cleanup is intended to dispatch remotely.

Keep the boundaries visible​

  • Use annotated, module-level functions for ordinary @task declarations so transform_function_to_interface and the default resolver can do their work.
  • Call task objects with keyword inputs; compilation produces promises/nodes, while local execution unwraps values and later recreates promises.
  • Provide a cache version when using the direct legacy cache settings, and do not mix those settings with a Cache object.
  • Treat enable_deck as the current option; disable_deck is deprecated and cannot be supplied with it.
  • Supply node_dependency_hints only to dynamic tasks.
  • Do not pass dynamic or eager tasks to array-node map wrappers that require default execution mode.
  • For hosted execution, provide serialization settings with the project, domain, version, and image configuration needed by PythonAutoContainerTask to build its command and image.