prefect.task_engine

Functions

run_task_sync ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1647" 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#L1678" 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#L1709" 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#L1768" 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#L1854" 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#L226" 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#L231" 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#L268" 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#L409" 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#L259" 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#L365" 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#L371" 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#L358" 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#L254" 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#L426" 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#L793" 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#L510" 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#L473" 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#L1020" 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#L436" 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#L431" 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#L739" 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#L709" 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#L654" 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#L616" 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#L724" 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#L811" 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#L601" 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#L984" 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#L550" 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#L749" 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#L914" 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#L957" 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#L891" 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#L1037" 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#L1420" 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#L1120" 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#L1083" 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#L1631" 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#L1047" 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#L1042" 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#L1367" 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#L1335" 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#L1279" 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#L1242" 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#L1350" 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#L1438" 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#L1227" 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#L1611" 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#L1174" 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#L1378" 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#L1540" 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#L1585" 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#L1517" 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.