prefect.task_engine

Functions

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

run_task_sync(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Union[R, State, None]

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

run_task_async(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Union[R, State, None]

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

run_generator_task_sync(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Generator[R, None, None]

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

run_generator_task_async(task: 'Task[P, R]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> AsyncGenerator[R, None]

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

run_task(task: 'Task[P, Union[R, Coroutine[Any, Any, R]]]', task_run_id: Optional[UUID] = None, task_run: Optional[TaskRun] = None, parameters: Optional[dict[str, Any]] = None, wait_for: Optional['OneOrManyFutureOrResult[Any]'] = None, return_type: Literal['state', 'result'] = 'result', dependencies: Optional[dict[str, set[RunInput]]] = None, context: Optional[dict[str, Any]] = None) -> Union[R, State, None, Coroutine[Any, Any, Union[R, State, None]]]

Runs the provided task.

Args:

• task: The task to run
• task_run_id: The ID of the task run; if not provided, a new task run will be created
• task_run: The task run object; if not provided, a new task run will be created
• parameters: The parameters to pass to the task
• wait_for: A list of futures to wait for before running the task
• return_type: The return type to return; either "state" or "result"
• dependencies: A dictionary of task run inputs to use for dependency tracking
• context: A dictionary containing the context to use for the task run; only required if the task is running on in a remote environment

Returns:

• The result of the task run

Classes

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

Raised when a task run exceeds its timeout.

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

Methods:

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

compute_transaction_key(self) -> Optional[str]

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

handle_rollback(self, txn: Transaction) -> None

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

is_cancelled(self) -> bool

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

is_running(self) -> bool

Whether or not the engine is currently running a task.

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

log_finished_message(self) -> None

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

record_terminal_state_timing(self, state: State) -> None

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

state(self) -> State

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

Methods:

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

asset_context(self)

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

begin_run(self) -> None

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

call_hooks(self, state: Optional[State] = None) -> None

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

call_task_fn(self, transaction: Transaction) -> Union[ResultRecord[Any], None, Coroutine[Any, Any, R], R]

Convenience method to call the task function. Returns a coroutine if the task is async.

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

can_retry(self, exc_or_state: Exception | State[R]) -> bool

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

client(self) -> SyncPrefectClient

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

handle_crash(self, exc: BaseException) -> None

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

handle_exception(self, exc: Exception) -> None

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

handle_retry(self, exc_or_state: Exception | State[R]) -> bool

Handle any task run retries.

• If the task has retries left, and the retry condition is met, set the task to retrying and return True.
• If the task has a retry delay, place in AwaitingRetry state with a delayed scheduled time.
• If the task has no retries left, or the retry condition is not met, return False.

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

handle_success(self, result: R, transaction: Transaction) -> Union[ResultRecord[R], None, Coroutine[Any, Any, R], R]

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

handle_timeout(self, exc: TimeoutError) -> None

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

initialize_run(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> Generator[Self, Any, Any]

Enters a client context and creates a task run if needed.

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

result(self, raise_on_failure: bool = True) -> 'Union[R, State, None]'

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

run_context(self)

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

set_state(self, state: State[R], force: bool = False) -> State[R]

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

setup_run_context(self, client: Optional[SyncPrefectClient] = None)

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

start(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> Generator[None, None, None]

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

transaction_context(self) -> Generator[Transaction, None, None]

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

wait_until_ready(self) -> None

Sync version: Waits until the scheduled time (if its the future), then enters Running.

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

Methods:

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

asset_context(self)

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

begin_run(self) -> None

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

call_hooks(self, state: Optional[State] = None) -> None

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

call_task_fn(self, transaction: AsyncTransaction) -> Union[ResultRecord[Any], None, Coroutine[Any, Any, R], R]

Convenience method to call the task function. Returns a coroutine if the task is async.

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

can_retry(self, exc_or_state: Exception | State[R]) -> bool

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

client(self) -> PrefectClient

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

handle_crash(self, exc: BaseException) -> None

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

handle_exception(self, exc: Exception) -> None

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

handle_retry(self, exc_or_state: Exception | State[R]) -> bool

Handle any task run retries.

• If the task has retries left, and the retry condition is met, set the task to retrying and return True.
• If the task has a retry delay, place in AwaitingRetry state with a delayed scheduled time.
• If the task has no retries left, or the retry condition is not met, return False.

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

handle_success(self, result: R, transaction: AsyncTransaction) -> Union[ResultRecord[R], None, Coroutine[Any, Any, R], R]

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

handle_timeout(self, exc: TimeoutError) -> None

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

initialize_run(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> AsyncGenerator[Self, Any]

Enters a client context and creates a task run if needed.

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

result(self, raise_on_failure: bool = True) -> 'Union[R, State, None]'

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

run_context(self)

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

set_state(self, state: State, force: bool = False) -> State

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

setup_run_context(self, client: Optional[PrefectClient] = None)

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

start(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> AsyncGenerator[None, None]

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

transaction_context(self) -> AsyncGenerator[AsyncTransaction, None]

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

wait_until_ready(self) -> None

Waits until the scheduled time (if its the future), then enters Running.