Skip to main content

Python Function Tasks and Execution Behaviors

A Python function becomes a Flyte task when you decorate it with @task; the decorator normally creates a PythonFunctionTask in DEFAULT execution mode. The task extracts its input and output interface from the function's annotations, so the function itself is the source of both the task body and its typed interface.

The main inheritance relationships are:

PythonAutoContainerTask
├── PythonFunctionTask
│   └── AsyncPythonFunctionTask
│       └── EagerAsyncPythonFunctionTask
└── PythonInstanceTask

PythonFunctionTask is the function-backed branch. PythonInstanceTask is the branch for plugin tasks whose execution behavior is supplied by the task class rather than by a user callable. The ExecutionBehavior enum is defined inside PythonFunctionTask:

class ExecutionBehavior(Enum):
    DEFAULT = 1
    DYNAMIC = 2
    EAGER = 3

Standard tasks​

Use @task when the function should run as an ordinary Flyte task:

from flytekit import task

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

task() defaults its execution_mode argument to PythonFunctionTask.ExecutionBehavior.DEFAULT. During construction, PythonFunctionTask calls transform_function_to_interface with the function and its docstring, then removes any names supplied through ignore_input_vars. It also derives the task name with extract_task_module(task_function). The resulting task retains the callable as task_function and exposes the selected mode through execution_mode.

For a default task, dispatch is direct:

PythonFunctionTask.execute(**kwargs)
        │
        └── DEFAULT → self._task_function(**kwargs)

The execute() method in PythonFunctionTask checks the enum and invokes the wrapped function for DEFAULT. The task decorator also uses this machinery for function-based extensions: a plugin can subclass PythonFunctionTask and register its configuration and task class with TaskPlugins.register_pythontask_plugin().

The default resolver requires the function to be accessible at module level. Constructing a PythonFunctionTask around a nested or local function raises ValueError, except for nested functions in test modules whose names begin with test_. If a custom decorator wraps the function, use functools.wraps or functools.update_wrapper so Flyte can recognize the module-level function. Alternatively, supply a custom TaskResolverMixin when implementing a task with different loading behavior.

pickle_untyped=True is available when you need to pickle untyped outputs instead of declaring output types, but the constructor documents this as a convenience that is not recommended for production use.

Dynamic tasks with @dynamic​

Use @dynamic when the task's Python body must use runtime input values to construct a workflow. For example, this test creates a variable number of task calls with a normal Python range:

from flytekit import dynamic, task

@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

res = ranged_int_to_str(a=5)
assert res == ["fast-2", "fast-3", "fast-4", "fast-5", "fast-6"]

In flytekit/core/dynamic_workflow_task.py, dynamic is not a separate task implementation. It is a partial application of task.task with the execution mode fixed:

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

At runtime, PythonFunctionTask.execute() routes a dynamic task to dynamic_execute() rather than directly calling the user function. The execution path depends on the current ExecutionState:

local execution
    └── create/cache PythonFunctionWorkflow → execute it locally

TASK_EXECUTION on a backend worker
    └── compile_into_workflow() → DynamicJobSpec

LOCAL_TASK_EXECUTION
    └── call the task function with native values

For a backend task execution, compile_into_workflow() creates a compilation context with a d prefix, creates a PythonFunctionWorkflow around the task function, compiles that workflow, and serializes it. If the compiled workflow has nodes, the method collects its task templates and returns a _dynamic_job.DynamicJobSpec containing the tasks, nodes, outputs, and subworkflows. If it produces no nodes, it returns the strict outputs as a LiteralMap instead. ReferenceTask objects are explicitly rejected inside this process with ValueError because reference tasks are not supported within dynamic tasks.

Local dynamic execution uses the cached PythonFunctionWorkflow but keeps the function inputs as native Python values. The result is converted back into Flyte literals using translate_inputs_to_literals; a None result or VoidPromise becomes a VoidPromise for the dynamic task.

The dynamic workflow module warns that the generated workflow is processed like any other workflow and recommends keeping dynamic workflows under fifty tasks. A loop can otherwise produce a much larger graph than a manually authored workflow.

Registering dependencies that runtime logic hides​

Flyte can discover ordinary task dependencies while compiling a static graph. A dynamic body can instead choose a task, launch plan, or workflow based on runtime input. Pass those entities in node_dependency_hints:

@dynamic(node_dependency_hints=[t1, t2])
def dt(mode: int) -> int:
    if mode == 1:
        return t1()
    if mode == 2:
        return t2()
    raise ValueError("Invalid mode")

PythonFunctionTask stores the value in node_dependency_hints, but validates that it is only used with ExecutionBehavior.DYNAMIC. Supplying hints to a default task raises ValueError with the message that hints should only be used on dynamic tasks; for static tasks and workflows, Flyte can find dependencies automatically.

Async function tasks​

Decorating an async def function with @task produces an AsyncPythonFunctionTask. This subclass changes the call boundary to an async Flyte-entity handler:

async def __call__(self, *args, **kwargs):
    return await async_flyte_entity_call_handler(self, *args, **kwargs)

Its async_execute() method awaits the wrapped function in default mode:

AsyncPythonFunctionTask.__call__
    └── async_flyte_entity_call_handler
          └── async_execute()
                └── DEFAULT → await self._task_function(**kwargs)

The class assigns execute = loop_manager.synced(async_execute), giving synchronous task-dispatch code an entry point that runs the coroutine through Flytekit's loop manager. async_execute() does not implement dynamic execution: when its execution_mode is DYNAMIC, it raises NotImplementedError. Consequently, dynamic and async eager execution modes are distinct; an eager task can call a dynamic task, but an async function task cannot itself use dynamic mode through AsyncPythonFunctionTask.async_execute().

Eager workflows with @eager​

Use @eager for an async def function whose Flyte entity calls should become live executions instead of being compiled into one dynamic workflow specification:

from flytekit import eager, task

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

The eager() decorator constructs an EagerAsyncPythonFunctionTask and forces deck generation on. Its constructor removes any execution_mode supplied through kwargs, sets metadata.is_eager = True, and passes PythonFunctionTask.ExecutionBehavior.EAGER to the parent constructor. Thus, eager mode is not an option that can be overridden on an eager task.

Unlike dynamic execution, eager execution does not first compile the function into a workflow specification. The eager decorator's documentation describes each called Flyte entity as its own execution with a Flyte URL, with the result or error fetched when that execution finishes. On a real backend execution, EagerAsyncPythonFunctionTask.async_execute() calls run_with_backend(). That method changes the execution state to EAGER_EXECUTION, awaits the task function, renders the controller's execution information into an Eager Executions deck, and converts an EagerException into a FlyteNonRecoverableSystemException after rendering the deck.

The controller is the worker queue for these child executions. If an eager task is running remotely without a queue in the context, execute() obtains a remote from the configured Flytekit plugin, derives project and domain from the execution context, constructs a Controller, installs its SIGINT and SIGTERM handlers, and runs async_execute() with that controller in context. Nested eager executions track their root through EAGER_ROOT_ENV_NAME; the source uses the _F_EE_ROOT environment setting when choosing the root tag.

Local eager execution takes a different branch. async_execute() detects a local execution, switches the context to EAGER_LOCAL_EXECUTION, and awaits the original task function directly. This makes the normal Python async patterns usable locally. The tests exercise nested eager tasks and parallel scheduling with asyncio.gather:

@eager
async def parent_wf(a: int, b: int) -> typing.Tuple[int, int]:
    t1 = asyncio.create_task(base_wf(x=a))
    t2 = asyncio.create_task(base_wf(x=b))
    i1, i2 = await asyncio.gather(t1, t2)
    return i1, i2

get_as_workflow() provides the serialized workflow boundary for an eager task. It creates an ImperativeWorkflow, adds the eager task with its typed inputs and outputs, and attaches an EagerFailureHandlerTask as the failure handler. The handler lists executions tagged with execution_tag.key == "eager-exec" that are still UNDEFINED, QUEUED, or RUNNING, then terminates them. EagerFailureTaskResolver can rehydrate that handler without the original eager task's Python interface.

For a non-sandbox cluster using FlyteRemote and client_credentials authentication, the @eager documentation requires client_secret_group and client_secret_key; the decorator documentation notes that these are not needed for a sandbox cluster started with flytectl demo start.

Instance tasks for plugin authors​

Choose PythonInstanceTask when a task has no user-defined function body. It is an abstract subclass of PythonAutoContainerTask; the platform-specific task class supplies execute() instead. The class documentation shows the intended shape: define an instance at module level and invoke that instance like a task. The module-level definition lets Flyte's module loader capture the module and variable needed to load the task class automatically.

A concrete plugin example is DBTRun in plugins/flytekit-dbt/flytekitplugins/dbt/task.py. It declares its interface explicitly and overrides execute():

class DBTRun(PythonInstanceTask):
    def __init__(self, name: str, **kwargs):
        super(DBTRun, self).__init__(
            task_type="dbt-run",
            name=name,
            task_config=None,
            interface=Interface(
                inputs=kwtypes(input=DBTRunInput),
                outputs=kwtypes(output=DBTRunOutput),
            ),
            **kwargs,
        )

    def execute(self, **kwargs) -> DBTRunOutput:
        ...

The corresponding usage is an instance variable, not a decorated function:

dbt_run_task = DBTRun(name="test-task")

@workflow
def my_workflow() -> DBTRunOutput:
    return dbt_run_task(
        input=DBTRunInput(
            project_dir="tests/jaffle_shop",
            profiles_dir="tests/jaffle_shop/profiles",
            profile="jaffle_shop",
        )
    )

Flytekit uses the same base for other platform-defined tasks, including ShellTask, NotebookTask, DBTRun, and DuckDBQuery. Contrast that with a function-backed plugin such as MMCloudTask: it subclasses PythonFunctionTask, retains a user function, and registers its configuration and task class through TaskPlugins.register_pythontask_plugin().

Constraints and mode boundaries​

  • Keep default task functions at module level. Nested and local functions fail the default task-resolver check at construction time; preserve wrapper metadata with functools.wraps or functools.update_wrapper when using custom decorators.
  • Set node_dependency_hints only on @dynamic tasks. PythonFunctionTask rejects hints for DEFAULT and other non-dynamic modes.
  • Do not combine async dynamic execution with AsyncPythonFunctionTask: its async_execute() raises NotImplementedError for DYNAMIC.
  • Eager tasks always use EAGER; EagerAsyncPythonFunctionTask removes an incoming execution_mode argument and replaces it with eager metadata.
  • Map-task integrations accept default-mode PythonFunctionTask instances or PythonInstanceTask instances, not dynamic or eager tasks.
  • Treat dynamic graph size as a practical constraint. flytekit/core/dynamic_workflow_task.py recommends fewer than fifty generated tasks and points large-scale identical runs toward map tasks.
  • Keep ReferenceTask out of dynamic task bodies. Dynamic compilation raises ValueError when it encounters one while collecting serialized task templates.
  • For eager execution, distinguish local async execution from backend execution: local mode runs the coroutine with the eager-local execution state, while backend mode needs the controller/remote path and, where applicable, the FlyteRemote client-credentials configuration.