prefect.task_engine

Functions

run_task_sync ^{<a href="https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_engine.py#L1654" 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#L1686" 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#L1718" 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#L1779" 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#L1867" 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#L229" 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#L234" 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#L271" 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#L412" 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#L262" 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#L368" 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#L374" 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#L361" 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#L257" 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#L429" 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#L795" 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#L513" 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#L476" 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#L1024" 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#L439" 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#L434" 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#L742" 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#L712" 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#L657" 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#L619" 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#L727" 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#L813" 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#L604" 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#L988" 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#L553" 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#L752" 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#L917" 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#L961" 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#L894" 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#L1041" 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#L1425" 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#L1124" 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#L1087" 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#L1638" 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#L1051" 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#L1046" 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#L1371" 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#L1339" 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#L1283" 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#L1246" 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#L1354" 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#L1443" 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#L1231" 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#L1618" 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#L1178" 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#L1382" 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#L1546" 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#L1592" 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#L1523" 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.