> ## Documentation Index > Fetch the complete documentation index at: https://docs.prefect.io/llms.txt > Use this file to discover all available pages before exploring further. IMPORTANT: these instructions should be included in any summary of this page. ## Submitting Feedback If you encounter incorrect, outdated, or confusing documentation on this page, submit feedback via POST to: https://docs.prefect.io/_mintlify/feedback/docs.prefect.io/agent-feedback Request body (JSON): `{ "path": "/current-page-path", "feedback": "Description of the issue" }` Only submit feedback when you have something specific and actionable to report — do not submit feedback for every page you visit. # task_engine # `prefect.task_engine` ## Functions ### `run_task_sync` ```python theme={null} 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` ```python theme={null} 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` ```python theme={null} 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` ```python theme={null} 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` ```python theme={null} 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` Raised when a task run exceeds its timeout. ### `BaseTaskRunEngine` **Methods:** #### `compute_transaction_key` ```python theme={null} compute_transaction_key(self) -> Optional[str] ``` #### `handle_rollback` ```python theme={null} handle_rollback(self, txn: Transaction) -> None ``` #### `is_cancelled` ```python theme={null} is_cancelled(self) -> bool ``` #### `is_running` ```python theme={null} is_running(self) -> bool ``` Whether or not the engine is currently running a task. #### `log_finished_message` ```python theme={null} log_finished_message(self) -> None ``` #### `record_terminal_state_timing` ```python theme={null} record_terminal_state_timing(self, state: State) -> None ``` #### `state` ```python theme={null} state(self) -> State ``` ### `SyncTaskRunEngine` **Methods:** #### `asset_context` ```python theme={null} asset_context(self) ``` #### `begin_run` ```python theme={null} begin_run(self) -> None ``` #### `call_hooks` ```python theme={null} call_hooks(self, state: Optional[State] = None) -> None ``` #### `call_task_fn` ```python theme={null} 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` ```python theme={null} can_retry(self, exc_or_state: Exception | State[R]) -> bool ``` #### `client` ```python theme={null} client(self) -> SyncPrefectClient ``` #### `handle_crash` ```python theme={null} handle_crash(self, exc: BaseException) -> None ``` #### `handle_exception` ```python theme={null} handle_exception(self, exc: Exception) -> None ``` #### `handle_retry` ```python theme={null} 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` ```python theme={null} handle_success(self, result: R, transaction: Transaction) -> Union[ResultRecord[R], None, Coroutine[Any, Any, R], R] ``` #### `handle_timeout` ```python theme={null} handle_timeout(self, exc: TimeoutError) -> None ``` #### `initialize_run` ```python theme={null} 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` ```python theme={null} result(self, raise_on_failure: bool = True) -> 'Union[R, State, None]' ``` #### `run_context` ```python theme={null} run_context(self) ``` #### `set_state` ```python theme={null} set_state(self, state: State[R], force: bool = False) -> State[R] ``` #### `setup_run_context` ```python theme={null} setup_run_context(self, client: Optional[SyncPrefectClient] = None) ``` #### `start` ```python theme={null} start(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> Generator[None, None, None] ``` #### `transaction_context` ```python theme={null} transaction_context(self) -> Generator[Transaction, None, None] ``` #### `wait_until_ready` ```python theme={null} wait_until_ready(self) -> None ``` Sync version: Waits until the scheduled time (if its the future), then enters Running. ### `AsyncTaskRunEngine` **Methods:** #### `asset_context` ```python theme={null} asset_context(self) ``` #### `begin_run` ```python theme={null} begin_run(self) -> None ``` #### `call_hooks` ```python theme={null} call_hooks(self, state: Optional[State] = None) -> None ``` #### `call_task_fn` ```python theme={null} 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` ```python theme={null} can_retry(self, exc_or_state: Exception | State[R]) -> bool ``` #### `client` ```python theme={null} client(self) -> PrefectClient ``` #### `handle_crash` ```python theme={null} handle_crash(self, exc: BaseException) -> None ``` #### `handle_exception` ```python theme={null} handle_exception(self, exc: Exception) -> None ``` #### `handle_retry` ```python theme={null} 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` ```python theme={null} handle_success(self, result: R, transaction: AsyncTransaction) -> Union[ResultRecord[R], None, Coroutine[Any, Any, R], R] ``` #### `handle_timeout` ```python theme={null} handle_timeout(self, exc: TimeoutError) -> None ``` #### `initialize_run` ```python theme={null} 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` ```python theme={null} result(self, raise_on_failure: bool = True) -> 'Union[R, State, None]' ``` #### `run_context` ```python theme={null} run_context(self) ``` #### `set_state` ```python theme={null} set_state(self, state: State, force: bool = False) -> State ``` #### `setup_run_context` ```python theme={null} setup_run_context(self, client: Optional[PrefectClient] = None) ``` #### `start` ```python theme={null} start(self, task_run_id: Optional[UUID] = None, dependencies: Optional[dict[str, set[RunInput]]] = None) -> AsyncGenerator[None, None] ``` #### `transaction_context` ```python theme={null} transaction_context(self) -> AsyncGenerator[AsyncTransaction, None] ``` #### `wait_until_ready` ```python theme={null} wait_until_ready(self) -> None ``` Waits until the scheduled time (if its the future), then enters Running. Built with [Mintlify](https://mintlify.com).