temporalio.worker.Replayer

class documentation

class Replayer: (source)

Constructor: Replayer(workflows, workflow_task_executor, workflow_runner, unsandboxed_workflow_runner, ...)

Replayer to replay workflows from history.

Method	`__init__`	Create a replayer to replay workflows from history.
Method	`config`	Config, as a dictionary, used to create this replayer.
Async Method	`replay_workflow`	Replay a workflow for the given history.
Async Method	`replay_workflows`	Replay workflows for the given histories.
Async Method	`workflow_replay_iterator`	Replay workflows for the given histories.
Instance Variable	`_config`	Undocumented

def __init__(self, *, workflows: Sequence[type], workflow_task_executor: concurrent.futures.ThreadPoolExecutor | None = None, workflow_runner: WorkflowRunner = SandboxedWorkflowRunner(), unsandboxed_workflow_runner: WorkflowRunner = UnsandboxedWorkflowRunner(), namespace: str = 'ReplayNamespace', data_converter: temporalio.converter.DataConverter = temporalio.converter.DataConverter.default, interceptors: Sequence[Interceptor] = [], build_id: str | None = None, identity: str | None = None, workflow_failure_exception_types: Sequence[type[BaseException]] = [], debug_mode: bool = False, runtime: temporalio.runtime.Runtime | None = None, disable_safe_workflow_eviction: bool = False): (source) ¶

Create a replayer to replay workflows from history.

See temporalio.worker.Worker.__init__ for a description of most of the arguments. Most of the same arguments need to be passed to the replayer that were passed to the worker when the workflow originally ran.

Note, unlike the worker, for the replayer the workflow_task_executor will default to a new thread pool executor with no max_workers set that will be shared across all replay calls and never explicitly shut down. Users are encouraged to provide their own if needing more control.

def config(self) -> ReplayerConfig: (source) ¶

Config, as a dictionary, used to create this replayer.

Returns
`ReplayerConfig`	Configuration, shallow-copied.

async def replay_workflow(self, history: temporalio.client.WorkflowHistory, *, raise_on_replay_failure: bool = True) -> WorkflowReplayResult: (source) ¶

Replay a workflow for the given history.

Parameters
history:`temporalio.client.WorkflowHistory`	The history to replay. Can be fetched directly, or use `temporalio.client.WorkflowHistory.from_json` to parse a history downloaded via `tctl` or the web UI.
raise_on_replay_failure:`bool`	If `True` (the default), this will raise a `WorkflowReplayResult.replay_failure` if it is present.
Returns
`WorkflowReplayResult`	Undocumented

async def replay_workflows(self, histories: AsyncIterator[temporalio.client.WorkflowHistory], *, raise_on_replay_failure: bool = True) -> WorkflowReplayResults: (source) ¶

Replay workflows for the given histories.

This is a shortcut for workflow_replay_iterator that iterates all results and aggregates information about them.

Parameters
histories:`AsyncIterator[temporalio.client.WorkflowHistory]`	The histories to replay, from an async iterator.
raise_on_replay_failure:`bool`	If `True` (the default), this will raise the first replay failure seen.
Returns
`WorkflowReplayResults`	Aggregated results.

@asynccontextmanager
async def workflow_replay_iterator(self, histories: AsyncIterator[temporalio.client.WorkflowHistory]) -> AsyncIterator[AsyncIterator[WorkflowReplayResult]]: (source) ¶

Replay workflows for the given histories.

This is a context manager for use via async with. The value is an iterator for use via async for.

Parameters
histories:`AsyncIterator[temporalio.client.WorkflowHistory]`	The histories to replay, from an async iterator.
Returns
`AsyncIterator[AsyncIterator[WorkflowReplayResult]]`	An async iterator that returns replayed workflow results as they are replayed.

_config = (source) ¶

Undocumented