-
Notifications
You must be signed in to change notification settings - Fork 370
Expand file tree
/
Copy pathexample_basic.py
More file actions
40 lines (30 loc) · 1.31 KB
/
example_basic.py
File metadata and controls
40 lines (30 loc) · 1.31 KB
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
35
36
37
38
39
40
import asyncio
from datetime import timedelta
from temporalio.api.workflowservice.v1 import ListNamespacesRequest
from temporalio.client import Client
from testcontainers.temporal import TemporalContainer
async def main():
with TemporalContainer() as temporal:
print(f"Temporal gRPC address: {temporal.get_grpc_address()}")
print(f"Temporal Web UI: {temporal.get_web_ui_url()}")
# Connect a Temporal client
client = await Client.connect(temporal.get_grpc_address())
# List available namespaces
resp = await client.service_client.workflow_service.list_namespaces(ListNamespacesRequest())
for ns in resp.namespaces:
print(f"Namespace: {ns.namespace_info.name}")
# Start a workflow (untyped — no workflow definition class needed)
handle = await client.start_workflow(
"GreetingWorkflow",
id="greeting-wf-1",
task_queue="greeting-queue",
execution_timeout=timedelta(seconds=10),
memo={"env": "example"},
)
print(f"Started workflow: {handle.id}")
# Describe the workflow
desc = await handle.describe()
print(f"Workflow type: {desc.workflow_type}")
print(f"Task queue: {desc.task_queue}")
if __name__ == "__main__":
asyncio.run(main())