Skip to main content

Task authoring and execution

Declaring a task​

When a Python function needs to become a Flyte node, annotate its inputs and return value and apply @task. The decorator constructs the task object from that function, so the function name, module, and typed interface are available to Flyte without a separate interface declaration.

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

For a plugin-backed task, pass the plugin's configuration object and task metadata through the decorator:

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

task.py builds a TaskMetadata instance from options such as cache, cache_version, retries, interruptible, deprecated, and timeout. It then finds a Python-task plugin with TaskPlugins.find_pythontask_plugin(type(task_config)). If no specialized plugin is registered, the normal implementation is PythonFunctionTask; coroutine functions are switched to AsyncPythonFunctionTask when the selected plugin is the ordinary PythonFunctionTask. Finally, update_wrapper(task_instance, decorated_fn) preserves the decorated function's metadata on the task object.

The public decorator also forwards container and execution settings, including container_image, environment, requests, limits, resources, secret_requests, task_resolver, enable_deck, deck_fields, and pod_template. Configure resources instead of combining it with requests or limits, and provide Secret objects in secret_requests as required by the decorator's validation.

Interfaces from annotations and type maps​

PythonFunctionTask calls transform_function_to_interface with the function and a Docstring. Its native Interface therefore contains the function's annotated inputs and outputs, while its Flyte-facing interface is produced by transforming that interface into a typed interface. A function's docstring is also used by PythonTask to populate Documentation: its short description and long description are copied into the task documentation, or merged into explicitly supplied documentation.

For interfaces that are assembled programmatically, kwtypes returns an ordered mapping from names to Python types. The same helper is used for reference interfaces and annotations such as structured-data columns:

superset_cols = kwtypes(Name=str, Age=int, Height=int)
subset_cols = kwtypes(Name=str)

@task
def t1() -> Annotated[pd.DataFrame, superset_cols]:
    return superset_df

@task
def t2(df: Annotated[pd.DataFrame, subset_cols]) -> Annotated[pd.DataFrame, subset_cols]:
    return df

@task
def t3(df: FlyteSchema[superset_cols]) -> FlyteSchema[superset_cols]:
    return df

The helper preserves keyword order, which is significant when the resulting map is used as interface and typing metadata.

The task abstraction layers​

Flyte's task hierarchy separates the Flyte model from Python-native execution:

Task
└── PythonTask
    └── PythonAutoContainerTask
        ├── PythonFunctionTask
        └── PythonInstanceTask

Task in base_task.py is close to the FlyteIDL TaskTemplate. Its constructor stores the task type, name, typed interface, TaskMetadata, task-type version, security context, and documentation. It also appends every constructed task to FlyteEntities.entities. The base class exposes interface, metadata, name, task_type, python_interface, security_context, and docs; a task with no Python-native interface returns None from python_interface.

PythonTask adds a Python Interface, optional plugin task_config, environment variables, and deck configuration. It converts the Python interface to a typed Flyte interface, provides Python input/output type lookup, and implements compile by calling create_and_link_node. This is the extension point used not only by function tasks, but also by container, SQL, map, and other task types.

PythonTask disables decks by default. Set enable_deck=True to activate deck generation, and select the generated sections with deck_fields. The default field tuple contains SOURCE_CODE, DEPENDENCIES, TIMELINE, INPUT, and OUTPUT, but those fields are only retained when decks are enabled. disable_deck is the deprecated spelling; passing both disable_deck and enable_deck raises ValueError, and an invalid deck_fields member also raises ValueError.

Metadata and validation​

Use TaskMetadata for execution properties that are part of the task model:

metadata = TaskMetadata(
    retries=1,
    timeout=datetime.timedelta(minutes=5),
    interruptible=True,
)

TaskMetadata.__post_init__ converts an integer timeout to datetime.timedelta(seconds=timeout). A non-int, non-timedelta timeout raises ValueError. Caching has explicit validation: cache=True requires a non-empty cache_version; cache_serialize=True requires caching; and cache_ignore_input_vars requires caching as well. The public @task path has a Cache API for cache configuration and should be preferred over manually creating an inconsistent metadata combination.

retry_strategy turns retries into Flyte's literal RetryStrategy. to_taskmetadata_model creates the Flyte task model and includes the SDK runtime metadata, timeout, retries, interruptibility, cache discovery/version/serialization settings, ignored cache inputs, deprecation message, deck generation, pod-template name, and eager flag.

From a call to execution​

Calling a task object invokes Task.__call__, which delegates to flyte_entity_call_handler. During workflow compilation, PythonTask.compile creates and links a node. During local execution, Task.local_execute accepts either native values or Promise values, translates them into a LiteralMap, and calls sandbox_execute.

The local execution path is:

Task.local_execute
  -> translate_inputs_to_literals
  -> optional LocalTaskCache lookup
  -> Task.sandbox_execute
  -> Task.dispatch_execute
  -> output Literals wrapped as Promise objects

If caching is enabled in LocalConfig and local caching is enabled, local_execute looks up the task name, cache version, literal inputs, and ignored input names in LocalTaskCache. A cache overwrite skips the lookup. On a miss, the task runs and the resulting literal map is stored. Regardless of caching, a task with no declared outputs returns VoidPromise; otherwise the declared output names are paired with Promise objects. A mismatch between the number of declared outputs and returned literals raises AssertionError.

PythonTask.dispatch_execute supplies the Python-specific middle of this path. It calls pre_execute first, converts the input literal map with TypeEngine.literal_map_to_kwargs, invokes execute(**native_inputs), then calls post_execute. Unless overridden, pre_execute returns the existing execution parameters and post_execute returns the result unchanged. Output values are converted back with TypeEngine.async_to_literal and returned in a LiteralMap.

Output packaging follows the declared interface. Zero outputs become an empty literal map; one output is mapped directly, with a special case for a one-element NamedTuple; multiple outputs are indexed against the declared output names. A tuple supplied as the value of an individual output is rejected. Conversion errors identify the task and output position. IgnoreOutputs is an exception marker for tasks whose outputs may safely be ignored, such as distributed or peer-to-peer algorithms; PythonTask.dispatch_execute allows that exception to propagate to the caller layer.

Local and hosted execution expose different error wrapping. During local execution, input conversion and user-function errors are re-raised with the task name added to the message. During hosted execution, input conversion and output-conversion failures are wrapped as FlyteNonRecoverableSystemException, while user-function failures are wrapped as FlyteUserRuntimeException.

Rehydrating Python tasks in a container​

A hosted Python task must identify the task object again when pyflyte-execute starts. TaskResolverMixin defines the serialization protocol: an implementation supplies location, name, load_task(loader_args), loader_args(settings, task), and get_all_tasks. Its optional task_name hook defaults to None.

PythonAutoContainerTask chooses default_task_resolver when no resolver is supplied. Its command builder in python_auto_container.py emits the resolver location and resolver-specific loader arguments after the pyflyte-execute arguments:

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

For the default resolver, those loader arguments identify the task module and task name. The resolver imports the module and retrieves the named task variable. This is why an ordinary PythonFunctionTask must generally wrap a module-level function: with the default resolver, PythonFunctionTask rejects nested or local functions. Test functions are allowed, as are module-level functions wrapped with functools.wraps or functools.update_wrapper. For another loading strategy, implement all abstract methods of TaskResolverMixin or pass a resolver such as a class-storage resolver.

Use PythonInstanceTask when there is no user-defined function body and a platform-defined execute implementation should run instead. It is an abstract wrapper over PythonAutoContainerTask; a subclass can be instantiated and called like this pattern from its class documentation:

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

The instance's declared interface determines the accepted arguments, and the subclass supplies the execution behavior.

Python function execution​

PythonFunctionTask stores the original callable in task_function. In its default mode, execute simply calls that function:

if self.execution_mode == self.ExecutionBehavior.DEFAULT:
    return self._task_function(**kwargs)

The constructor requires a callable, derives its native interface, removes any names listed in ignore_input_vars from the exposed interface, and derives the task name from the function's module and name. ignore_input_vars changes the task interface; it does not remove the parameter from the underlying callable, so those values must be supplied by the surrounding execution arrangement.

The task supports three execution modes: DEFAULT, DYNAMIC, and EAGER. Normal tasks use DEFAULT. Coroutine functions selected by @task use AsyncPythonFunctionTask, whose asynchronous call handler awaits the function; dynamic execution is not supported by that class and raises NotImplementedError if selected.

Dynamic tasks​

Use dynamic when the function body must create a workflow from runtime-native inputs. The public decorator is an alias for task.task with 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

PythonFunctionTask.execute routes this mode to dynamic_execute. Locally, dynamic_execute creates or reuses a PythonFunctionWorkflow and executes it with native values. In backend task execution, compile_into_workflow compiles that generated workflow and returns a DynamicJobSpec containing the generated nodes, task templates, outputs, and subworkflows. A generated workflow with no nodes returns a LiteralMap containing its strict outputs instead.

Set node_dependency_hints only for dynamic tasks. Supplying it to a normal task raises ValueError, because static tasks and workflows allow Flyte to discover their dependencies automatically. Dynamic compilation also rejects ReferenceTask dependencies, and the generated runtime workflow can become large as the function creates more nodes.

Async and eager execution​

For an ordinary asynchronous task, define an async function under @task and await the task call. AsyncPythonFunctionTask.__call__ uses async_flyte_entity_call_handler; its async_execute awaits the underlying function in default mode, while execute is a synchronized bridge for the normal dispatch path.

Use @eager when the asynchronous function should act as an eager workflow and invoke other Flyte entities:

from flytekit import task, eager

@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 forces ExecutionBehavior.EAGER, sets TaskMetadata.is_eager=True, and enables decks by default. Locally it switches execution to EAGER_LOCAL_EXECUTION and awaits the function. For backend execution it creates a Controller and worker queue, installs SIGINT and SIGTERM handlers, and runs in EAGER_EXECUTION mode. A backend eager call requires a user-space execution_id; the nonlocal setup also rejects an already-present worker queue.

run_with_backend renders the eager call stack into a Deck. If an EagerException occurs, it renders the Deck and then raises FlyteNonRecoverableSystemException. Eager root propagation uses _F_EE_ROOT; if it is absent, the current execution name is used as the root tag, and the worker queue propagates the value to downstream calls. run(remote, ss, ...) provides a local-testing helper that points the eager parent at a remote cluster.

get_as_workflow wraps an eager task in an ImperativeWorkflow and attaches an EagerFailureHandlerTask as its failure handler. That cleanup task is dispatch-only: it queries the configured remote for active executions tagged eager-exec, terminates them until none remain, and its execute method raises AssertionError if called directly.

Integration constraints and common pitfalls​

  • Map implementations consume these task abstractions selectively. ArrayNodeMapTask accepts a default-mode PythonFunctionTask or a PythonInstanceTask, rejects dynamic/eager function tasks, and currently permits at most one output. The legacy map_task wrapper accepts PythonFunctionTask, PythonInstanceTask, or functools.partial, propagates task metadata and security context, and supports node overrides such as with_overrides.
  • Enabling caching manually with TaskMetadata(cache=True) without a cache version fails during construction. The decorator's Cache handling should be used for public task declarations.
  • Deck fields do not enable decks by themselves. Pass enable_deck=True; disable_deck is deprecated. Eager tasks are the exception because EagerAsyncPythonFunctionTask enables them by default for eager execution rendering.
  • Keep normal function tasks accessible to the resolver at module scope. For nested functions, provide a resolver that can load them or use a task type with an appropriate TaskResolverMixin implementation.
  • A task instance is registered globally when Task.__init__ runs. This is part of task construction, not a later registration call, and is relevant to code that inspects FlyteEntities.entities during translation.