prefect.task_runners

Classes

TaskRunner ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L63" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Abstract base class for task runners.

A task runner is responsible for submitting tasks to the task run engine running in an execution environment. Submitted tasks are non-blocking and return a future object that can be used to wait for the task to complete and retrieve the result.

Task runners are context managers and should be used in a with block to ensure proper cleanup of resources.

Methods:

duplicate ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L85" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

duplicate(self) -> Self

Return a new instance of this task runner with the same configuration.

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L118" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any | unmapped[Any] | allow_failure[Any]], wait_for: Iterable[PrefectFuture[R]] | None = None) -> PrefectFutureList[F]

Submit multiple tasks to the task run engine.

Args:

• task: The task to submit.
• parameters: The parameters to use when running the task.
• wait_for: A list of futures that the task depends on.

Returns:

• An iterable of future objects that can be used to wait for the tasks to
• complete and retrieve the results.

name ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L80" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

name(self) -> str

The name of this task runner

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L91" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> F

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L101" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> F

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L110" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> F

ThreadPoolTaskRunner ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L242" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

A task runner that executes tasks in a separate thread pool.

Attributes:

• max_workers: The maximum number of threads to use for executing tasks. Defaults to PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS or sys.maxsize.

Examples:

Use a thread pool task runner with a flow:

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def some_io_bound_task(x: int) -> int:
    # making a query to a database, reading a file, etc.
    return x * 2

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3)) # use at most 3 threads at a time
def my_io_bound_flow():
    futures = []
    for i in range(10):
        future = some_io_bound_task.submit(i * 100)
        futures.append(future)

    return [future.result() for future in futures]

Use a thread pool task runner as a context manager:

from prefect.task_runners import ThreadPoolTaskRunner

@task
def some_io_bound_task(x: int) -> int:
    # making a query to a database, reading a file, etc.
    return x * 2

# Use the runner directly
with ThreadPoolTaskRunner(max_workers=2) as runner:
    future1 = runner.submit(some_io_bound_task, {"x": 1})
    future2 = runner.submit(some_io_bound_task, {"x": 2})

    result1 = future1.result()  # 2
    result2 = future2.result()  # 4

Configure max workers via settings:

# Set via environment variable
# export PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS=8

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner())  # Uses 8 workers from setting
def my_flow():
    ...

Methods:

cancel_all ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L506" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

cancel_all(self) -> None

duplicate ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L338" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

duplicate(self) -> 'ThreadPoolTaskRunner[R]'

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L483" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L491" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L498" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L342" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L351" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L359" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]

Submit a task to the task run engine running in a separate thread.

Args:

• task: The task to submit.
• parameters: The parameters to use when running the task.
• wait_for: A list of futures that the task depends on.

Returns:

• A future object that can be used to wait for the task to complete and
• retrieve the result.

ProcessPoolTaskRunner ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L749" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

A task runner that executes tasks in a separate process pool.

This task runner uses ProcessPoolExecutor to run tasks in separate processes, providing true parallelism for CPU-bound tasks and process isolation. Tasks are executed with proper context propagation and error handling.

Attributes:

• max_workers: The maximum number of processes to use for executing tasks. Defaults to multiprocessing.cpu_count() if PREFECT_TASKS_RUNNER_PROCESS_POOL_MAX_WORKERS is not set.
• initializer: An optional picklable callable run once in each worker subprocess before it executes any tasks, mirroring ProcessPoolExecutor's initializer.
• initargs: A tuple of arguments passed to initializer.

Examples:

Use a process pool task runner with a flow:

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task
def compute_heavy_task(n: int) -> int:
    # CPU-intensive computation that benefits from process isolation
    return sum(i ** 2 for i in range(n))

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def my_flow():
    futures = []
    for i in range(10):
        future = compute_heavy_task.submit(i * 1000)
        futures.append(future)

    return [future.result() for future in futures]

Use a process pool task runner as a context manager:

from prefect.task_runners import ProcessPoolTaskRunner

@task
def my_task(x: int) -> int:
    return x * 2

# Use the runner directly
with ProcessPoolTaskRunner(max_workers=2) as runner:
    future1 = runner.submit(my_task, {"x": 1})
    future2 = runner.submit(my_task, {"x": 2})

    result1 = future1.result()  # 2
    result2 = future2.result()  # 4

Configure max workers via settings:

# Set via environment variable
# export PREFECT_TASKS_RUNNER_PROCESS_POOL_MAX_WORKERS=8

from prefect import flow
from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner())  # Uses 8 workers from setting
def my_flow():
    ...

Run custom setup in each worker subprocess:

import sys
from prefect.task_runners import ProcessPoolTaskRunner

def add_to_path(*paths):
    sys.path[:0] = paths

runner = ProcessPoolTaskRunner(
    initializer=add_to_path,
    initargs=("/opt/my_app/libs",),
)

Methods:

cancel_all ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1263" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

cancel_all(self) -> None

duplicate ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L877" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

duplicate(self) -> Self

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1240" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1248" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1255" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]

set_subprocess_message_processor_factories ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L908" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

set_subprocess_message_processor_factories(self, subprocess_message_processor_factories: Iterable[_SubprocessMessageProcessorFactory] | None = None) -> None

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1140" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1149" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1157" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]

Submit a task to the task run engine running in a separate process.

Args:

• task: The task to submit.
• parameters: The parameters to use when running the task.
• wait_for: A list of futures that the task depends on.
• dependencies: A dictionary of dependencies for the task.

Returns:

• A future object that can be used to wait for the task to complete and
• retrieve the result.

subprocess_message_processor_factories ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L888" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

subprocess_message_processor_factories(self) -> tuple[_SubprocessMessageProcessorFactory, ...]

subprocess_message_processor_factories ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L894" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

subprocess_message_processor_factories(self, subprocess_message_processor_factories: Iterable[_SubprocessMessageProcessorFactory] | None = None) -> None

PrefectTaskRunner ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1337" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

Methods:

duplicate ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1341" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

duplicate(self) -> 'PrefectTaskRunner[R]'

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1400" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectDistributedFuture[R]]

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1408" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectDistributedFuture[R]]

map ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1415" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectDistributedFuture[R]]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1345" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectDistributedFuture[R]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1354" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectDistributedFuture[R]

submit ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1362" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectDistributedFuture[R]

Submit a task to the task run engine running in a separate thread.

Args:

• task: The task to submit.
• parameters: The parameters to use when running the task.
• wait_for: A list of futures that the task depends on.

Returns:

• A future object that can be used to wait for the task to complete and
• retrieve the result.