Skip to main content

Task authoring and execution

Declare a task with @task​

Use @task when a Python function should become a Flyte task. The function annotations provide the task interface, while decorator arguments configure execution metadata and the task plugin:

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

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

The public construction path is implemented in flytekit/core/task.py. It creates a TaskMetadata object from options such as cache, cache_version, retries, interruptible, deprecated, and timeout. It then finds a Python-task plugin for task_config, selects an asynchronous plugin for coroutine functions when appropriate, and forwards task settings—including container_image, environment, resources, task_resolver, execution mode, node dependency hints, and Deck settings—to the task instance. Finally, it applies functools.update_wrapper to the instance.

For a normal function, the resulting object is a PythonFunctionTask. The function name and module, as well as the input and output interface, are inferred from the callable:

@task
def my_func(a: int) -> str:
    ...

PythonFunctionTask calls transform_function_to_interface with the function and its docstring. It removes any names supplied through ignore_input_vars from the Flyte interface, derives the task name with extract_task_module, and retains the original callable in task_function. A missing task_function raises ValueError.

Configure execution metadata​

TaskMetadata is the common metadata object carried by Task. Its fields include retries, timeout, interruptibility, deprecation text, caching, pod-template selection, Deck generation, and eager-task identification. Integer timeout values are interpreted as seconds and converted to datetime.timedelta values. For example, these cache combinations are validated when the metadata is constructed:

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

cache=True without a non-empty cache_version raises ValueError. cache_serialize=True and cache_ignore_input_vars are also rejected unless caching is enabled. TaskMetadata.retry_strategy returns a Flyte RetryStrategy, and to_taskmetadata_model() serializes the metadata into Flyte's task model, including the Flyte SDK runtime metadata.

During local execution, caching is used only when both TaskMetadata.cache and LocalConfig.auto().cache_enabled are true. LocalConfig.cache_overwrite bypasses an existing entry. Cache keys use the task name, cache version, translated input literal map, and cache_ignore_input_vars.

Understand the task abstraction ladder​

Flyte task authoring is layered:

Task
  └─ PythonTask
       └─ PythonAutoContainerTask
            └─ PythonFunctionTask

Each layer adds a different responsibility.

Task: Flyte task identity and local dispatch​

base_task.Task is the lowest-level abstraction and is close to the FlyteIDL task template. It stores the task type, name, typed interface, metadata, task-type version, security context, and documentation. Constructing a Task also appends it to FlyteEntities.entities, so task construction has a global entity-registration side effect.

Task.__call__ routes invocation through flyte_entity_call_handler. Its compile() method is intentionally unimplemented at this level; concrete task types supply compilation behavior. The abstract extension points are pre_execute, execute, and dispatch_execute. The base class also exposes hooks for hosted containers, Kubernetes pods, SQL, custom serialized data, configuration, and extended resources; each default implementation returns None.

Task.local_execute() handles the local boundary between Python values and Flyte literals:

native values or Promises
        │
        ▼
translate_inputs_to_literals
        │
        ▼
sandbox_execute → dispatch_execute
        │
        ▼
LiteralMap
        │
        ▼
Promise / VoidPromise results

It translates incoming keyword arguments using the task interface, optionally consults LocalTaskCache, and calls sandbox_execute. sandbox_execute switches to a task sandbox execution state before calling dispatch_execute. Afterward, declared output names are paired with returned literals and wrapped as Promise objects. A task with no declared outputs returns VoidPromise; a mismatch between declared output count and returned literal count raises AssertionError.

PythonTask: typed Python interfaces and execution lifecycle​

base_task.PythonTask is for tasks with a Python-native Interface, including tasks whose execution method is supplied by a plugin rather than by a user function. It converts that interface to a typed Flyte interface, stores task_config and environment variables, and exposes Python input and output types through get_input_types, get_type_for_input_var, and get_type_for_output_var.

Unlike Task, PythonTask.compile() creates and links a workflow node with create_and_link_node. Its construct_node_metadata() carries the task name, timeout, retry strategy, and interruptibility into node metadata.

Its dispatch_execute() lifecycle is:

pre_execute(user parameters)
    → LiteralMap to native Python inputs
    → execute(**native_inputs)
    → post_execute(user parameters, result)
    → native outputs to LiteralMap

Input conversion uses TypeEngine.literal_map_to_kwargs. Output conversion uses TypeEngine.async_to_literal, maps values to the declared output names, and attaches tracked output metadata when present. In local execution, input and user-code exceptions are re-raised with task context; during remote execution, user-code failures are wrapped in FlyteUserRuntimeException, while conversion failures are wrapped according to the remote/local handling in PythonTask.dispatch_execute.

Override pre_execute when a task needs to modify execution parameters before inputs are converted, and override post_execute to clean up or alter the result. IgnoreOutputs is an exception marker for tasks whose outputs can safely be discarded; the dispatch lifecycle allows an IgnoreOutputs raised by post_execute to propagate to the caller layer.

Decks are disabled by default in PythonTask. Use enable_deck=True to enable them and deck_fields to select fields. disable_deck is retained as a deprecated alternative; supplying both disable_deck and enable_deck raises ValueError, and invalid deck fields are rejected. Supplying deck_fields without enabling Decks leaves the selected field list empty.

PythonAutoContainerTask: hosted execution and rehydration​

PythonFunctionTask inherits container behavior from PythonAutoContainerTask. When serialized for a hosted Flyte execution, get_default_command() builds a pyflyte-execute command containing input and output locations, checkpoint arguments, the resolver location, and resolver loader arguments:

def get_default_command(self, settings: SerializationSettings) -> List[str]:
    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),
    ]
    return container_args

The resolver is the bridge from the serialized command back to a Python task object. TaskResolverMixin requires implementations to provide location, name, load_task(loader_args), loader_args(settings, task), and get_all_tasks(). task_name() may provide a custom name. The default resolver records the task module and name and later imports the module to retrieve the task attribute. If you use a custom resolver, keep loader_args() and load_task() compatible and ensure the serialized module or storage identifier remains rehydratable.

Execute ordinary Python function tasks​

For a DEFAULT PythonFunctionTask, invocation eventually reaches PythonFunctionTask.execute, which calls the stored function:

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)

When the task is used while compiling a workflow, PythonTask.compile() creates a node and the call handler returns Promise values representing that node's outputs. When the task is locally executed, Task.local_execute() converts inputs to literals, invokes PythonTask.dispatch_execute(), and converts the function's native return value back to literals before returning promises or native local results through the call path.

PythonTask._output_to_literal_map() maps multiple returned values by declared output order. A single output has special handling for a one-element NamedTuple; a tuple returned for an individual output otherwise raises TypeError. Functions with no outputs produce an empty literal map and ultimately a VoidPromise in the base local execution path.

Use kwtypes when an interface must be declared programmatically rather than inferred from a function. It preserves keyword order while building an ordered mapping of names to types, as in this reference entity:

ref_entity = get_reference_entity(
    _identifier_model.ResourceType.WORKFLOW,
    "project",
    "dev",
    "my.other.workflow",
    "abc123",
    inputs=kwtypes(a=str, b=int),
    outputs={},
)

Choose a specialized execution mode​

PythonFunctionTask.ExecutionBehavior defines DEFAULT, DYNAMIC, and EAGER. The normal task decorator uses DEFAULT; specialized decorators and constructors select the other modes.

Async functions​

The task construction path detects coroutine functions. A plain coroutine function uses AsyncPythonFunctionTask when the selected plugin is PythonFunctionTask. Its __call__ awaits async_flyte_entity_call_handler, and async_execute awaits the stored function in DEFAULT mode. Async dynamic execution is explicitly unsupported and raises NotImplementedError.

Dynamic tasks​

flytekit/core/dynamic_workflow_task.py defines dynamic as a partial application of task.task with ExecutionBehavior.DYNAMIC:

dynamic = functools.partial(task.task, execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC)

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

Dynamic task code receives native inputs and can use them in Python control flow. In local execution, PythonFunctionTask.dynamic_execute() creates or reuses a PythonFunctionWorkflow, executes it with the native inputs, and returns a LiteralMap. In TASK_EXECUTION mode, it compiles the generated workflow into a DynamicJobSpec for the Flyte engine. In LOCAL_TASK_EXECUTION, it calls the function directly. Any other execution state raises ValueError.

Dynamic tasks may provide node_dependency_hints, but PythonFunctionTask rejects those hints unless the execution mode is DYNAMIC. During runtime compilation, reference tasks are unsupported inside the generated dynamic workflow, and serialized task entities must be TaskSpec models.

Eager tasks​

EagerAsyncPythonFunctionTask is the async eager-workflow implementation. Its constructor removes any supplied execution_mode, forces ExecutionBehavior.EAGER, sets TaskMetadata.is_eager=True, and enables Decks by default. A documented eager pattern calls ordinary tasks from an async eager function:

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

Locally, EagerAsyncPythonFunctionTask.async_execute() changes the execution state to EAGER_LOCAL_EXECUTION and awaits the user function. Remotely, it uses run_with_backend(): the eager controller provides a worker queue, nested Flyte entities become executions, and the current execution identifier supplies the tag. The _F_EE_ROOT environment variable propagates a root eager execution tag; when it is unset, the current execution name is used.

Remote eager entry requires a user-space execution ID. If a worker queue is not already present, the task constructs a default remote using the configured plugin and installs SIGINT/SIGTERM handlers. It renders the controller's execution view into an Eager Executions Deck. An EagerException is converted into FlyteNonRecoverableSystemException after the execution view is rendered.

get_as_workflow() wraps an eager task in an ImperativeWorkflow, wires the task's Python inputs and outputs, and adds an EagerFailureHandlerTask as the failure handler. That internal cleanup task polls Admin for executions tagged eager-exec in UNDEFINED, QUEUED, or RUNNING phases and terminates them. Its execute() method raises an assertion because cleanup is performed by its custom dispatch_execute() path.

Extend tasks and integrate with wrappers​

Use PythonInstanceTask when the task has a Python-native interface but no user-defined function body. It is an abstract base for PythonAutoContainerTask subclasses whose platform or plugin supplies execute:

x = MyInstanceTask(name="x", .....) 
x(a=5)  # depending on the interface of the defined task

Its constructor forwards name, task_config, task_type, an optional TaskResolverMixin, and container/task arguments to PythonAutoContainerTask. This makes the instance discoverable by the module and variable tracking used for task rehydration.

Map integrations deliberately constrain the function-task abstraction. ArrayNodeMapTask accepts PythonInstanceTask or PythonFunctionTask only when the function task uses DEFAULT execution, and it allows at most one output. MapPythonTask similarly accepts those task types and derives a collection-oriented interface. DYNAMIC and EAGER function tasks are therefore not interchangeable with these map wrappers.

For custom task types, subclass the appropriate layer and implement the required lifecycle methods. A plugin-backed task commonly supplies task_config, a typed Interface, pre_execute, execute, and any hosted representation hooks. A function-backed task should use PythonFunctionTask so annotation-based interface creation, resolver integration, output conversion, and execution-mode behavior remain consistent. If the default resolver cannot represent the callable—for example, a nested or local function—use a module-level function, preserve wrappers with functools.wraps or functools.update_wrapper, or supply a custom TaskResolverMixin.