forked from temporalio/samples-python
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathstarter.py
34 lines (27 loc) · 856 Bytes
/
starter.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import asyncio
from temporalio.client import Client
from custom_converter.shared import (
GreetingInput,
GreetingOutput,
greeting_data_converter,
)
from custom_converter.workflow import GreetingWorkflow
async def main():
# Connect client
client = await Client.connect(
"localhost:7233",
# Without this we get:
# TypeError: Object of type GreetingInput is not JSON serializable
data_converter=greeting_data_converter,
)
# Run workflow
result = await client.execute_workflow(
GreetingWorkflow.run,
GreetingInput("Temporal"),
id=f"custom_converter-workflow-id",
task_queue="custom_converter-task-queue",
)
assert isinstance(result, GreetingOutput)
print(f"Workflow result: {result.result}")
if __name__ == "__main__":
asyncio.run(main())