forked from temporalio/samples-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathreplayer.py
More file actions
22 lines (15 loc) · 671 Bytes
/
Copy pathreplayer.py
File metadata and controls
22 lines (15 loc) · 671 Bytes
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import asyncio
from temporalio.client import Client
from temporalio.worker import Replayer
from replay.worker import JustActivity, JustTimer, TimerThenActivity
async def main():
# Connect client
client = await Client.connect("localhost:7233")
# Fetch the histories of the workflows to be replayed
workflows = client.list_workflows('WorkflowId="replayer-workflow-id"')
histories = workflows.map_histories()
replayer = Replayer(workflows=[JustActivity, JustTimer, TimerThenActivity])
results = await replayer.replay_workflows(histories, raise_on_replay_failure=False)
print(results)
if __name__ == "__main__":
asyncio.run(main())