> ## 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. # flow_engine # `prefect.flow_engine` ## Functions ### `load_flow_run` ```python theme={null} load_flow_run(flow_run_id: UUID) -> FlowRun ``` ### `load_flow` ```python theme={null} load_flow(flow_run: FlowRun) -> Flow[..., Any] ``` ### `load_flow_and_flow_run` ```python theme={null} load_flow_and_flow_run(flow_run_id: UUID) -> tuple[FlowRun, Flow[..., Any]] ``` ### `send_heartbeats_sync` ```python theme={null} send_heartbeats_sync(engine: 'FlowRunEngine[Any, Any]') -> Generator[None, None, None] ``` ### `send_heartbeats_async` ```python theme={null} send_heartbeats_async(engine: 'AsyncFlowRunEngine[Any, Any]') -> AsyncGenerator[None, None] ``` ### `run_flow_sync` ```python theme={null} run_flow_sync(flow: Flow[P, R], flow_run: Optional[FlowRun] = None, parameters: Optional[Dict[str, Any]] = None, wait_for: Optional[Iterable[PrefectFuture[Any]]] = None, return_type: Literal['state', 'result'] = 'result', context: Optional[dict[str, Any]] = None) -> Union[R, State, None] ``` ### `run_flow_async` ```python theme={null} run_flow_async(flow: Flow[P, R], flow_run: Optional[FlowRun] = None, parameters: Optional[Dict[str, Any]] = None, wait_for: Optional[Iterable[PrefectFuture[Any]]] = None, return_type: Literal['state', 'result'] = 'result', context: Optional[dict[str, Any]] = None) -> Union[R, State, None] ``` ### `run_generator_flow_sync` ```python theme={null} run_generator_flow_sync(flow: Flow[P, R], flow_run: Optional[FlowRun] = None, parameters: Optional[Dict[str, Any]] = None, wait_for: Optional[Iterable[PrefectFuture[Any]]] = None, return_type: Literal['state', 'result'] = 'result', context: Optional[dict[str, Any]] = None) -> Generator[R, None, None] ``` ### `run_generator_flow_async` ```python theme={null} run_generator_flow_async(flow: Flow[P, R], flow_run: Optional[FlowRun] = None, parameters: Optional[Dict[str, Any]] = None, wait_for: Optional[Iterable[PrefectFuture[R]]] = None, return_type: Literal['state', 'result'] = 'result', context: Optional[dict[str, Any]] = None) -> AsyncGenerator[R, None] ``` ### `run_flow` ```python theme={null} run_flow(flow: Flow[P, R], flow_run: Optional[FlowRun] = None, parameters: Optional[Dict[str, Any]] = None, wait_for: Optional[Iterable[PrefectFuture[R]]] = None, return_type: Literal['state', 'result'] = 'result', error_logger: Optional[logging.Logger] = None, context: Optional[dict[str, Any]] = None) -> R | State | None | Coroutine[Any, Any, R | State | None] | Generator[R, None, None] | AsyncGenerator[R, None] ``` ### `run_flow_in_subprocess` ```python theme={null} run_flow_in_subprocess(flow: 'Flow[..., Any]', flow_run: 'FlowRun | None' = None, parameters: dict[str, Any] | None = None, wait_for: Iterable[PrefectFuture[Any]] | None = None, context: dict[str, Any] | None = None, env: dict[str, str] | None = None) -> multiprocessing.context.SpawnProcess ``` Run a flow in a subprocess. Note the result of the flow will only be accessible if the flow is configured to persist its result. **Args:** * `flow`: The flow to run. * `flow_run`: The flow run object containing run metadata. * `parameters`: The parameters to use when invoking the flow. * `wait_for`: The futures to wait for before starting the flow. * `context`: A serialized context to hydrate before running the flow. If not provided, the current context will be used. A serialized context should be provided if this function is called in a separate memory space from the parent run (e.g. in a subprocess or on another machine). * `env`: Additional environment variables to set in the subprocess. **Returns:** * A multiprocessing.context.SpawnProcess representing the process that is running the flow. ## Classes ### `FlowRunTimeoutError` Raised when a flow run exceeds its defined timeout. ### `BaseFlowRunEngine` **Methods:** #### `cancel_all_tasks` ```python theme={null} cancel_all_tasks(self) -> None ``` #### `heartbeat_seconds` ```python theme={null} heartbeat_seconds(self) -> Optional[int] ``` Get the heartbeat interval from settings. #### `is_pending` ```python theme={null} is_pending(self) -> bool ``` #### `is_running` ```python theme={null} is_running(self) -> bool ``` #### `state` ```python theme={null} state(self) -> State ``` ### `FlowRunEngine` **Methods:** #### `begin_run` ```python theme={null} begin_run(self) -> State ``` #### `call_flow_fn` ```python theme={null} call_flow_fn(self) -> Union[R, Coroutine[Any, Any, R]] ``` Convenience method to call the flow function. Returns a coroutine if the flow is async. #### `call_hooks` ```python theme={null} call_hooks(self, state: Optional[State] = None) -> None ``` #### `client` ```python theme={null} client(self) -> SyncPrefectClient ``` #### `create_flow_run` ```python theme={null} create_flow_run(self, client: SyncPrefectClient) -> FlowRun ``` #### `handle_crash` ```python theme={null} handle_crash(self, exc: BaseException) -> None ``` #### `handle_exception` ```python theme={null} handle_exception(self, exc: Exception, msg: Optional[str] = None, result_store: Optional[ResultStore] = None) -> State ``` #### `handle_success` ```python theme={null} handle_success(self, result: R) -> R ``` #### `handle_timeout` ```python theme={null} handle_timeout(self, exc: TimeoutError) -> None ``` #### `initialize_run` ```python theme={null} initialize_run(self) ``` Enters a client context and creates a flow run if needed. #### `load_subflow_run` ```python theme={null} load_subflow_run(self, parent_task_run: TaskRun, client: SyncPrefectClient, context: FlowRunContext) -> Union[FlowRun, None] ``` This method attempts to load an existing flow run for a subflow task run, if appropriate. If the parent task run is in a final but not COMPLETED state, and not being rerun, then we attempt to load an existing flow run instead of creating a new one. This will prevent the engine from running the subflow again. If no existing flow run is found, or if the subflow should be rerun, then no flow run is returned. #### `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[SyncPrefectClient] = None) ``` #### `start` ```python theme={null} start(self) -> Generator[None, None, None] ``` ### `AsyncFlowRunEngine` Async version of the flow run engine. NOTE: This has not been fully asyncified yet which may lead to async flows not being fully asyncified. **Methods:** #### `begin_run` ```python theme={null} begin_run(self) -> State ``` #### `call_flow_fn` ```python theme={null} call_flow_fn(self) -> Coroutine[Any, Any, R] ``` Convenience method to call the flow function. Returns a coroutine if the flow is async. #### `call_hooks` ```python theme={null} call_hooks(self, state: Optional[State] = None) -> None ``` #### `client` ```python theme={null} client(self) -> PrefectClient ``` #### `create_flow_run` ```python theme={null} create_flow_run(self, client: PrefectClient) -> FlowRun ``` #### `handle_crash` ```python theme={null} handle_crash(self, exc: BaseException) -> None ``` #### `handle_exception` ```python theme={null} handle_exception(self, exc: Exception, msg: Optional[str] = None, result_store: Optional[ResultStore] = None) -> State ``` #### `handle_success` ```python theme={null} handle_success(self, result: R) -> R ``` #### `handle_timeout` ```python theme={null} handle_timeout(self, exc: TimeoutError) -> None ``` #### `initialize_run` ```python theme={null} initialize_run(self) ``` Enters a client context and creates a flow run if needed. #### `load_subflow_run` ```python theme={null} load_subflow_run(self, parent_task_run: TaskRun, client: PrefectClient, context: FlowRunContext) -> Union[FlowRun, None] ``` This method attempts to load an existing flow run for a subflow task run, if appropriate. If the parent task run is in a final but not COMPLETED state, and not being rerun, then we attempt to load an existing flow run instead of creating a new one. This will prevent the engine from running the subflow again. If no existing flow run is found, or if the subflow should be rerun, then no flow run is returned. #### `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) -> AsyncGenerator[None, None] ``` Built with [Mintlify](https://mintlify.com).