flow_engine

prefect.flow_engine

Functions

load_flow_run

load_flow_run(flow_run_id: UUID) -> FlowRun

load_flow

load_flow(flow_run: FlowRun) -> Flow[..., Any]

load_flow_and_flow_run

load_flow_and_flow_run(flow_run_id: UUID) -> tuple[FlowRun, Flow[..., Any]]

send_heartbeats_sync

send_heartbeats_sync(engine: 'FlowRunEngine[Any, Any]') -> Generator[None, None, None]

send_heartbeats_async

send_heartbeats_async(engine: 'AsyncFlowRunEngine[Any, Any]') -> AsyncGenerator[None, None]

run_flow_sync

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

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

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

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

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

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 = 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

cancel_all_tasks(self) -> None

heartbeat_seconds

heartbeat_seconds(self) -> Optional[int]

Get the heartbeat interval from settings.

is_pending

is_pending(self) -> bool

is_running

is_running(self) -> bool

setup_flow_run_suspension_request

setup_flow_run_suspension_request(self) -> Generator[FlowRunSuspensionRequest, None, None]

state

state(self) -> State

FlowRunEngine

Methods:

begin_run

begin_run(self) -> State

call_flow_fn

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

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

client

client(self) -> SyncPrefectClient

create_flow_run

create_flow_run(self, client: SyncPrefectClient) -> FlowRun

handle_cancellation

handle_cancellation(self, exc: BaseException) -> None

Force this run through Cancelling -> Cancelled.

Used when capture_sigterm has determined (via the cancellation listener) that the SIGTERM was the runner asking for cancellation, not an unrelated termination. The same ownership rule governs the on_cancellation hook: if the engine owns cancellation handling, it must drive Cancelling -> Cancelled and run the hook locally; if an external runner owns cancellation, the child avoids duplicating state history and hooks. Engine-reported Crashed states and on_crashed hooks always remain engine-owned.

handle_crash

handle_crash(self, exc: BaseException) -> None

handle_exception

handle_exception(self, exc: Exception, msg: Optional[str] = None, result_store: Optional[ResultStore] = None) -> State

handle_success

handle_success(self, result: R) -> R

handle_timeout

handle_timeout(self, exc: TimeoutError) -> None

initialize_run

initialize_run(self)

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

load_subflow_run

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 state, we return the existing subflow run to avoid re-execution (unless the parent is being rerun and the subflow did not complete, in which case a fresh run is desired).

If the parent task run is in a non-final state (e.g. still Running after a process restart), we also look for an existing subflow run to reattach to, preventing duplicate subflow runs from being created under the same parent task run.

If no existing flow run is found, or if the subflow should be rerun, then no flow run is returned.

result

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

run_context

run_context(self)

set_state

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

setup_run_context

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

start

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

begin_run(self) -> State

call_flow_fn

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

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

client

client(self) -> PrefectClient

create_flow_run

create_flow_run(self, client: PrefectClient) -> FlowRun

handle_cancellation

handle_cancellation(self, exc: BaseException) -> None

Force this run through Cancelling -> Cancelled.

Async counterpart to FlowRunEngine.handle_cancellation. Shielded from asyncio cancellation so engine-owned transitions always reach the server.

handle_crash

handle_crash(self, exc: BaseException) -> None

handle_exception

handle_exception(self, exc: Exception, msg: Optional[str] = None, result_store: Optional[ResultStore] = None) -> State

handle_success

handle_success(self, result: R) -> R

handle_timeout

handle_timeout(self, exc: TimeoutError) -> None

initialize_run

initialize_run(self)

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

load_subflow_run

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 state, we return the existing subflow run to avoid re-execution (unless the parent is being rerun and the subflow did not complete, in which case a fresh run is desired).

If the parent task run is in a non-final state (e.g. still Running after a process restart), we also look for an existing subflow run to reattach to, preventing duplicate subflow runs from being created under the same parent task run.

If no existing flow run is found, or if the subflow should be rerun, then no flow run is returned.

result

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

run_context

run_context(self)

set_state

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

setup_run_context

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

start

start(self) -> AsyncGenerator[None, None]