Skip to main content

Tasks and task execution

Define a task with a typed Python interface​

When you write a function task, you normally do not instantiate a task class yourself. Use @task; the decorator accepts the function and task settings, and produces a PythonFunctionTask (or a registered plugin subclass). The function's annotations become the task interface, so the same definition can be called from a workflow and executed locally:

from flytekit import task, workflow

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

@workflow
def my_workflow(x: int) -> int:
    return my_task(x=x)

This is the pattern used in docs/guides/1bb8844e-7f97-4d17-8c9f-f1e149f41a24_executing_workflows_and_tasks.md. A task call is routed through Task.__call__, which delegates to flyte_entity_call_handler. In a compiled workflow, PythonTask.compile calls create_and_link_node, so the call becomes a workflow node. In local execution, Task.local_execute accepts native values and Promises, translates them into a Flyte LiteralMap, dispatches the task, and wraps the results in Promise objects (or a VoidPromise for a task with no outputs).

The task hierarchy separates the Flyte-facing abstraction from Python-specific behavior:

Task
└── PythonTask
    └── PythonAutoContainerTask
        ├── PythonFunctionTask   (wraps a Python callable)
        └── PythonInstanceTask   (subclass supplies execute)

Task and PythonTask​

Task in flytekit/core/base_task.py is the root abstraction. Its constructor stores the task type, name, typed interface, TaskMetadata, task-type version, security context, and documentation. It also appends each instance to FlyteEntities.entities, so constructing a task has an entity-registration side effect used by serialization.

The base Task has no Python-native interface: python_interface returns None, and get_input_types returns None. Its default serialization hooks—get_container, get_k8s_pod, get_sql, get_custom, get_config, and get_extended_resources—return None. A task implementation supplies the relevant hooks and implements dispatch_execute, pre_execute, and execute.

PythonTask is the base for task types with a Python-native Interface, but without necessarily having a user function. Its constructor accepts task_type, name, task_config, interface, and an optional environment. It converts the Python interface to Flyte's typed interface and exposes the original interface through python_interface. It also provides get_input_types, get_type_for_input_var, and get_type_for_output_var, which the type engine uses during local dispatch.

Use a direct PythonTask subclass when the task type has its own execution implementation rather than a Python function body. The subclass must implement execute; the inherited dispatch_execute handles the Flyte literal boundary and lifecycle around it. task_config is retained for plugin-specific handling and is available through the task_config property.

Function-backed and instance-backed tasks​

PythonFunctionTask is the usual extension point for a callable. Its constructor calls transform_function_to_interface with the function and its docstring, removes any ignore_input_vars, and derives the task name from the function's module. For example, the decorator form above supplies the annotated x: int -> int interface automatically. pickle_untyped=True is also supported for untyped outputs, although the constructor explicitly describes that option as not recommended for production.

PythonInstanceTask is the alternative when there is no user-defined function body. It is an abstract PythonAutoContainerTask; the subclass provides the platform-defined execute method, while the instance carries its interface and task configuration. The class docstring describes the intended shape as an instantiated task object that is called with its inputs, for example x(a=5). This is useful for plugin-style tasks whose execution behavior belongs to the task class rather than to a decorated function.

Both forms are auto-container tasks. PythonFunctionTask is appropriate when the task's behavior is a callable; PythonInstanceTask is appropriate when a subclass defines the behavior. A plugin can also register a PythonFunctionTask derivative with TaskPlugins.register_pythontask_plugin; the factory uses the registered class for that plugin's configuration type and otherwise falls back to PythonFunctionTask.

Understand local execution and dispatch​

A local call does more than invoke the Python function directly. Task.local_execute first translates keyword arguments—including native constants, Promises, and collections containing them—using translate_inputs_to_literals and the task's interface. It then creates a LiteralMap and calls sandbox_execute. The sandbox adds task-sandbox user parameters to the execution context and calls dispatch_execute.

The normal dispatch path in PythonTask is:

input LiteralMap
    ↓
pre_execute(user_params)
    ↓
LiteralMap → Python-native inputs
    ↓
execute(**native_inputs)
    ↓
post_execute(user_params, native_outputs)
    ↓
Python-native outputs → output LiteralMap
    ↓
optional Deck writing

pre_execute runs before input conversion. The default implementation returns the existing user parameters, but a subclass can return modified parameters; the docstring specifically calls out using this stage to set up user-process context before type transformers run. execute is the required task operation. post_execute runs after it and defaults to a no-op, so subclasses can clean up or alter the returned value before output conversion.

PythonTask.dispatch_execute creates a child execution context containing the parameters returned by pre_execute, converts the input literals with TypeEngine.literal_map_to_kwargs, and invokes execute. It converts ordinary native outputs asynchronously with TypeEngine.async_to_literal. A LiteralMap or DynamicJobSpec returned by execution is passed through instead of being converted again; this is how dynamic tasks return a runtime job description or an already-created local literal map.

The error behavior distinguishes local and hosted execution. During local execution, input-conversion and user-code errors are re-raised with an annotation such as Error encountered while executing '<task name>'. During hosted execution, user-code errors are wrapped in FlyteUserRuntimeException, while other output-conversion errors become FlyteNonRecoverableSystemException. Upload and download exceptions are re-raised in either path.

After dispatch, Task.local_execute checks that the number of output literals matches the declared interface. A no-output task returns VoidPromise(self.name); otherwise, each output literal is wrapped in a Promise and passed to create_task_output. This is why a local workflow composition can continue to use task outputs as Promise-like values even though the task body has already run.

The output boundary also has a few concrete edge cases. PythonTask._output_to_literal_map gives declared output names to returned values, handles the single-output NamedTuple convention, and rejects a tuple supplied as an individual output. Conversion failures include the task name, output position or name, and expected Python type in the raised error.

Lifecycle hooks and Decks​

Deck generation is disabled by default in PythonTask. Enable it with enable_deck=True and select fields with deck_fields:

@task(
    enable_deck=True,
    deck_fields=(DeckField.TIMELINE, DeckField.INPUT, DeckField.OUTPUT),
)
def summarize(value: str) -> str:
    return value.upper()

PythonTask rejects setting both disable_deck and enable_deck, warns that disable_deck is deprecated, and validates every deck_fields value against DeckField. When enabled, dispatch_execute adds the timeline deck to user parameters when requested, and _write_decks renders selected inputs and outputs with the type engine. PythonFunctionTask extends that behavior for source-code and dependency decks. For local execution, deck output is written at the end of _write_decks; hosted execution emits the decks through the runtime path.

Configure task metadata​

Pass execution policy through the decorator or through a TaskMetadata instance. The decorator's public arguments include cache, retries, interruptible, deprecated, and timeout:

from flytekit import task

@task(
    cache=True,
    cache_version="1.0",
    retries=2,
    timeout=60,
    interruptible=True,
)
def process(value: str) -> str:
    return value.strip()

TaskMetadata in flytekit/core/base_task.py stores these values along with cache_serialize, cache_ignore_input_vars, pod_template_name, generates_deck, and is_eager. Its validation is intentionally strict:

  • cache=True requires a non-empty cache_version.
  • cache_serialize=True requires cache=True.
  • cache_ignore_input_vars requires cache=True.
  • An integer timeout is interpreted as seconds and converted to datetime.timedelta; a non-integer value must already be a datetime.timedelta.

For example, the repository tests verify the valid cache combination and both invalid combinations:

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

assert foo.metadata.cache is True
assert foo.metadata.cache_serialize is True
assert foo.metadata.cache_version == "1.0"

with pytest.raises(ValueError):
    @task(cache=True)
    def foo_missing_cache_version(i: str):
        print(f"{i}")

with pytest.raises(ValueError):
    @task(cache_serialize=True)
    def foo_missing_cache(i: str):
        print(f"{i}")

TaskMetadata.retry_strategy converts retries into a Flyte RetryStrategy. to_taskmetadata_model maps caching and its version, retries, timeout, interruptibility, deprecation message, ignored cache inputs, pod-template name, Deck generation, and eager status into the Flyte IDL model, together with the Flyte SDK runtime metadata. PythonTask.construct_node_metadata separately places timeout, retry strategy, and interruptibility on the node metadata used when compiling a task call.

Local cache behavior​

Metadata caching affects local execution only when both the task requests caching and LocalConfig.auto().cache_enabled is true. Task.local_execute uses the translated input LiteralMap, task name, cache_version, and cache_ignore_input_vars to query LocalTaskCache. A cache hit skips sandbox_execute; a miss runs the task and stores the output literal map. Setting LocalConfig.cache_overwrite bypasses an existing entry and executes the task again before replacing the cached value.

The local cache therefore sits outside the lifecycle rather than inside execute:

native/Promise inputs
    ↓ translate_inputs_to_literals
input LiteralMap
    ↓ LocalTaskCache.get (when enabled)
cache hit → output LiteralMap
cache miss → sandbox_execute → LocalTaskCache.set

For mapped tasks, metadata is passed to the mapped-task constructor and applies to mapped instances. The repository uses this form with TaskMetadata(retries=1):

@task
def my_mappable_task(a: int) -> typing.Optional[str]:
    return str(a)

@workflow
def my_wf(x: typing.List[int]) -> typing.List[typing.Optional[str]]:
    return map_task(
        my_mappable_task,
        metadata=TaskMetadata(retries=1),
        concurrency=10,
        min_success_ratio=0.75,
    )(a=x).with_overrides(requests=Resources(cpu="10M"))

Serialize Python tasks into containers​

PythonAutoContainerTask supplies the hosted-execution representation for its subclasses. Give a function task an image, environment, resources, or a pod template through @task, or pass the same settings to a custom auto-container subclass. The task's environment is combined with SerializationSettings.env; duplicate task keys overwrite the settings keys. The container image can be a string or ImageSpec, and SerializationSettings.image_config resolves it to the registerable image.

The test suite demonstrates the environment merge with direct construction:

def foo(i: int):
    pass

settings = SerializationSettings(
    project="p",
    domain="d",
    version="v",
    image_config=cfg,
    env={"FOO": "bar"},
)
pytask = PythonFunctionTask(None, foo, None, environment={"BAZ": "baz"})
c = pytask.get_container(settings)
assert c.image == "xyz.com/abc:tag1"
assert c.env == {"FOO": "bar", "BAZ": "baz"}

With no command override, get_default_command creates a pyflyte-execute command containing input and output prefixes, checkpoint arguments, the resolver location, and resolver loader arguments. For the test task named task in module tests.flytekit.unit.core.test_python_auto_container, the generated tail is:

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

PythonAutoContainerTask.get_command calls the current command function. Use set_command_fn for a serialization-time command override and reset_command_fn to restore the default. Use set_resolver when the default resolver is not appropriate, such as a task that needs a custom loading strategy.

Resource configuration has two mutually exclusive forms. Supply requests and limits separately, or supply consolidated resources; PythonAutoContainerTask raises ValueError if resources is combined with either of the other two. secret_requests must contain only flytekit.Secret objects, otherwise construction raises AssertionError. An ImageSpec with runtime packages adds those packages to the container environment under RUNTIME_PACKAGES_ENV_NAME. Accelerators and shared memory are serialized through get_extended_resources.

A pod template changes the serialized shape. When pod_template is set, get_container returns None; get_k8s_pod instead serializes a pod whose primary container is merged with the generated image, resources, environment, and pyflyte-execute arguments. Labels and annotations from the PodTemplate are retained. pod_template_name is assigned to TaskMetadata.pod_template_name in the auto-container constructor and therefore overwrites a value already present in the metadata object.

Dynamic execution and resolver constraints​

PythonFunctionTask.ExecutionBehavior defines DEFAULT, DYNAMIC, and EAGER modes. In DEFAULT, execute calls the wrapped function directly. In DYNAMIC, it calls dynamic_execute, which treats the function body as a workflow. During local execution, dynamic_execute creates or reuses a PythonFunctionWorkflow and executes it locally, returning a LiteralMap. During hosted task execution, it compiles that workflow at runtime through compile_into_workflow and returns a DynamicJobSpec. A LOCAL_TASK_EXECUTION branch calls the function directly, while an unsupported execution state raises ValueError.

node_dependency_hints are accepted by PythonFunctionTask only when execution_mode is DYNAMIC; supplying them to a static task raises ValueError, because static task and workflow dependencies are discovered during normal compilation. The dynamic compiler also rejects ReferenceTask objects inside the generated dynamic workflow.

Finally, the default resolver requires a module-loadable task. PythonFunctionTask rejects nested or local functions when using default_task_resolver, except for test functions. Functions decorated by custom wrappers must preserve their module-level identity with functools.wraps or functools.update_wrapper, or you must provide a TaskResolverMixin. This constraint follows directly from the generated task-module and task-name loader arguments: hosted execution imports the module and resolves the named task rather than serializing an arbitrary local function body.

Map-task support follows the task shape as well: the repository's map-task implementation accepts PythonFunctionTask and PythonInstanceTask, with PythonFunctionTask restricted to ExecutionBehavior.DEFAULT. Use dynamic mode for a task that must construct a runtime workflow, not for a mapped task that is expected to execute the ordinary callable over its list inputs.