temporal_hello_world_workflow_activity_worker_quickstart.py

1import asyncio
2from datetime import timedelta
3from temporalio import workflow, activity
4from temporalio.client import Client
5from temporalio.worker import Worker
6
7# Step 1: Define the Activity
8@activity.define
9async def say_hello(name: str) -> str:
10    return f"Hello, {name}!"
11
12# Step 2: Define the Workflow
13@workflow.def_init
14class SayHelloWorkflow:
15    @workflow.run
16    async def run(self, name: str) -> str:
17        return await workflow.execute_activity(
18            say_hello, name, start_to_close_timeout=timedelta(seconds=5)
19        )
20
21async def main():
22    # Step 3: Connect to the Temporal server
23    client = await Client.connect("localhost:7233")
24
25    # Step 4: Run the Worker
26    # In a real app, the worker and the starter would be in separate files
27    async with Worker(
28        client,
29        task_queue="hello-task-queue",
30        workflows=[SayHelloWorkflow],
31        activities=[say_hello],
32    ):
33        # Step 5: Execute the Workflow
34        result = await client.execute_workflow(
35            SayHelloWorkflow.run,
36            "Temporal",
37            id="hello-workflow-id",
38            task_queue="hello-task-queue",
39        )
40
41        print(f"Result: {result}")
42
43if __name__ == "__main__":
44    asyncio.run(main())