class documentation

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: Optional[concurrent.futures.ThreadPoolExecutor] = 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: Optional[str] = None, identity: Optional[str] = None, workflow_failure_exception_types: Sequence[Type[BaseException]] = [], debug_mode: bool = False, runtime: Optional[temporalio.runtime.Runtime] = 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
ReplayerConfigConfiguration, 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.WorkflowHistoryThe 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:boolIf True (the default), this will raise a WorkflowReplayResult.replay_failure if it is present.
Returns
WorkflowReplayResultUndocumented
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:boolIf True (the default), this will raise the first replay failure seen.
Returns
WorkflowReplayResultsAggregated 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.

Undocumented