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. The same arguments need to be passed to the replayer that were passed to the worker when the workflow originally ran.

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