Skip to main content

Task authoring and execution

Task hierarchy: from an IDL task to a Python function​

A Flyte task has two distinct jobs: describe an executable entity to Flyte, and provide a Python-facing way to invoke it. The hierarchy in flytekit.core.base_task separates those responsibilities:

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

Task is the lowest-level abstraction and is closest to the FlyteIDL TaskTemplate. Its constructor accepts a task_type, name, typed IDL interface, optional TaskMetadata, security context, and documentation. Construction also appends the task to FlyteEntities.entities, which makes it available to Flyte's entity serialization machinery. A concrete task must implement dispatch_execute, pre_execute, and execute.

PythonTask adds a native Interface, plugin task_config, environment variables, and deck settings. It also supplies the standard Python execution lifecycle and converts between Flyte LiteralMap values and Python values. PythonAutoContainerTask adds container and resource configuration plus task resolution; PythonFunctionTask and PythonInstanceTask build on that container-oriented behavior.

In normal application code, use @task. The decorator creates a Python function task from the function's annotations:

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

In this example, PythonFunctionTask derives the input and output interface from the annotations. Inside my_wf, task calls produce bindings between nodes (y is the output promise from t1); during local execution, the workflow returns native values, so my_wf(a=5, b="hello ") returns (7, "hello world").

Configuring a task​

Pass execution and container options through @task. The task decorator's plugin lookup and task construction ultimately place these values on TaskMetadata or PythonAutoContainerTask:

@task(
    cache=True,
    cache_serialize=True,
    cache_version="1.0",
    retries=2,
    timeout=60,
    interruptible=True,
    container_image="repo/image:0.0.0",
    environment={"MODE": "production"},
)
def process(i: str) -> str:
    return i.upper()

requests, limits, or the combined resources option configure resource allocation on the auto-container task. Do not combine resources with requests or limits; the task configuration rejects that combination. Other auto-container options include secret_requests, pod_template, pod_template_name, and accelerator-related settings. When a task-specific environment is supplied, it is merged with the environment in SerializationSettings when the container or pod is serialized.

Deck generation is configured with enable_deck and optionally deck_fields. PythonTask defaults to decks disabled. disable_deck is retained for compatibility but is deprecated; setting both disable_deck and enable_deck raises ValueError.

Metadata, caching, and execution policy​

TaskMetadata in flytekit.core.base_task carries task-level execution policy. Its fields include retries, timeout, interruptible, deprecated, cache settings, pod_template_name, generates_deck, and is_eager.

The cache settings are validated when the metadata object is created. A cache version is required whenever caching is enabled, and serialization or ignored cache inputs are only valid with caching enabled:

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

@task(cache=True) raises ValueError because no cache version was supplied. Likewise, @task(cache_serialize=True) raises ValueError because cache=True is absent. cache_ignore_input_vars follows the same rule. The cache_serialize and cache_ignore_input_vars values are passed into the local cache lookup by Task.local_execute.

You can also attach metadata directly when a task is not created from a Python function. For example, a SQL task in the test suite is configured as follows:

sql = SQLTask(
    "my-query",
    query_template="SELECT * FROM hive.city.fact_airport_sessions WHERE ds = ... LIMIT 10",
    inputs=kwtypes(ds=datetime.datetime),
    outputs=kwtypes(results=FlyteSchema),
    metadata=TaskMetadata(retries=2, cache=True, cache_version="0.1"),
)

TaskMetadata(timeout=60) converts the integer seconds to datetime.timedelta(seconds=60). A timeout must otherwise be a datetime.timedelta or an integer; other values raise ValueError. to_taskmetadata_model() maps this configuration to Flyte's task model, including retry strategy, timeout, cache discovery/version, interruptibility, deprecation text, deck generation, and eager state.

For local execution, Task.local_execute translates native arguments and promises into a LiteralMap. If local caching is enabled, it checks LocalTaskCache using the task name, cache version, literal inputs, and ignored input names. On a miss it calls sandbox_execute, stores the resulting literal map, and converts outputs back to Promise objects. A task with no declared outputs returns VoidPromise.

The dispatch lifecycle​

At runtime, PythonTask.dispatch_execute is the boundary between Flyte literals and user Python code. The flow is:

input LiteralMap
    -> pre_execute(user parameters)
    -> TypeEngine.literal_map_to_kwargs(...)
    -> execute(**native_inputs)
    -> post_execute(...)
    -> TypeEngine.async_to_literal(...)
    -> output LiteralMap

pre_execute runs before input conversion, so subclasses can establish user-space context needed by type transformers. The default implementation returns the parameters unchanged. PythonTask.dispatch_execute then calls _literal_map_to_python_input, invokes execute, calls post_execute, and uses _output_to_literal_map to map native return values onto the declared output names. Output conversion is asynchronous internally and uses TypeEngine.async_to_literal for each output. A subclass can alter or clean up the result in post_execute.

PythonFunctionTask.execute invokes the captured function in default mode:

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

The dispatch path preserves original exceptions for local execution but wraps user exceptions as FlyteUserRuntimeException for remote execution. Input conversion failures are handled similarly, with local errors retaining a task-specific message and remote failures being classified as system failures.

If a task raises IgnoreOutputs, flytekit propagates that signal through execution to the entrypoint, which returns without writing outputs.pb. This is used by distributed-task implementations when a worker is allowed to run the computation but its outputs should not be uploaded—for example, non-chief workers in distributed training.

Python function execution modes​

PythonFunctionTask.ExecutionBehavior has three values: DEFAULT, DYNAMIC, and EAGER.

Default tasks​

The default mode runs the captured function as ordinary task user code. The task's interface is generated by transform_function_to_interface, and its name is derived from the function's module and qualified name by extract_task_module.

Dynamic tasks​

@dynamic creates a PythonFunctionTask with ExecutionBehavior.DYNAMIC. Use it when the task's Python body must inspect runtime inputs to decide which task calls to create:

@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

For local execution, dynamic_execute runs the generated PythonFunctionWorkflow with native inputs. During a real task execution, compile_into_workflow compiles that function body into a workflow specification and returns a DynamicJobSpec; Flyte then executes the generated nodes. Thus range(a) can use the runtime value, rather than being expanded during static workflow compilation.

If Flyte cannot discover a dynamic task's dependencies statically—such as conditionally referenced launch plans—pass node_dependency_hints. PythonFunctionTask rejects these hints on non-dynamic tasks. Dynamic expansion is materialized as workflow nodes, so large expansions should use the project's map_task facility instead; the dynamic-task tests and implementation treat dynamic workflows as ordinary workflow specifications after compilation.

Eager tasks​

Eager execution uses Python as the orchestrator while each task invocation can become a Flyte execution. EagerAsyncPythonFunctionTask forces TaskMetadata.is_eager=True, defaults enable_deck=True, and uses ExecutionBehavior.EAGER. During a remote execution, it creates a Controller worker queue, installs SIGINT and SIGTERM handlers, and runs the async task function. Calls made from the eager function are dispatched through that queue.

The controller and signal handlers are constructed on the main thread. This is a runtime requirement of EagerAsyncPythonFunctionTask.execute; a worker queue already present in the context is rejected for the entrypoint path.

get_as_workflow() exposes an eager task through an ImperativeWorkflow. It creates an EagerFailureHandlerTask as the workflow's failure handler. On remote execution, that handler finds child executions tagged with the eager parent execution and terminates executions still in UNDEFINED, QUEUED, or RUNNING phases.

Async Python functions​

Decorating an async def function selects AsyncPythonFunctionTask. Its __call__ awaits async_flyte_entity_call_handler, and its async_execute awaits the captured function; the synchronous execute entry point is generated with loop_manager.synced.

Async function tasks use the default execution behavior for normal async execution. AsyncPythonFunctionTask.async_execute explicitly raises NotImplementedError for DYNAMIC, and the class documentation states that eager and dynamic execution do not mix. Use EagerAsyncPythonFunctionTask for eager workflows rather than combining dynamic mode with an async task.

Resolving a task inside its container​

Serialization must tell pyflyte-execute which Python object to rehydrate. TaskResolverMixin defines that contract with location, loader_args, load_task, and get_all_tasks. The standard DefaultTaskResolver uses the task's module and name. The resulting command has the following shape:

pyflyte-execute --inputs s3://path/inputs.pb --output-prefix s3://outputs/location \
  --raw-output-data-prefix /tmp/data \
  --resolver flytekit.core.python_auto_container.default_task_resolver \
  -- \
  task-module repo_root.workflows.example task-name t1

At load time, the resolver imports repo_root.workflows.example and retrieves t1 with the supplied arguments. This is why a regular PythonFunctionTask must refer to a module-level function. With the default resolver, constructing a task around a nested or local function raises ValueError (test functions and functions preserved with functools.wraps/functools.update_wrapper are the documented exceptions). If a task cannot use module-and-name resolution, implement TaskResolverMixin and provide a resolver through task_resolver.

Extending task behavior​

Plugin-backed Python functions​

The decorator can select a specialized PythonFunctionTask based on the type of task_config. The Ray plugin demonstrates the extension contract:

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:
        ray.init(address=self._task_config.address)
        return user_params

The plugin registers the mapping with TaskPlugins.register_pythontask_plugin(RayJobConfig, RayFunctionTask). Thereafter, a task configured with RayJobConfig is instantiated as RayFunctionTask rather than the default PythonFunctionTask. The same subclass uses get_custom to serialize Ray's configuration into the task template's custom field.

Use pre_execute for setup that must happen before Flyte input conversion, and post_execute for cleanup or output adaptation. Override get_custom when a backend plugin needs additional serializable task data.

Tasks without a user function​

Use PythonInstanceTask when the platform-defined task class owns execution rather than a decorated function. It requires an explicit name and task_config, inherits PythonAutoContainerTask, and expects a subclass to implement execute. Shell and dbt task integrations use this pattern. The instance is still called like a task—for example, the class documentation shows an instance x being invoked as x(a=5)—but its identity is not derived from a function body.

Common failure modes​

  • Put ordinary Python function tasks at module scope. The default resolver cannot rehydrate nested or local functions.
  • Supply cache_version with cache=True, or use the cache configuration supported by the task decorator. cache_serialize and cache_ignore_input_vars do not enable caching by themselves.
  • Prefer enable_deck; disable_deck is deprecated and cannot be supplied together with enable_deck.
  • Set node_dependency_hints only on dynamic tasks.
  • Choose either the combined resources configuration or separate requests/limits; flytekit rejects mixing them.
  • Check decorator keyword names carefully. The task decorator rejects unrecognized arguments rather than silently adding them to the task.
  • Remember that ReferenceTask uses a remote task's signature and does not execute the referenced Python function body; its interface must match the remote task.