async handlers example

Author: dandavison

Diff for: message_passing/introduction/workflow_with_async_handlers.py

@@ -1,8 +1,11 @@
+import asyncio
 from dataclasses import dataclass
+from datetime import timedelta
 from enum import IntEnum
 from typing import Optional
 
-from temporalio import workflow
+from docutils import ApplicationError
+from temporalio import activity, workflow
 
 
 class Language(IntEnum):
@@ -11,6 +14,23 @@ class Language(IntEnum):
     FRENCH = 3
     SPANISH = 4
     PORTUGUESE = 5
+    ARABIC = 6
+    HINDI = 7
+
+
+@activity.defn
+async def call_greeting_service(to_language: Language) -> Optional[str]:
+    greetings = {
+        Language.CHINESE: "你好，世界",
+        Language.ENGLISH: "Hello, world",
+        Language.FRENCH: "Bonjour, monde",
+        Language.SPANISH: "¡Hola mundo",
+        Language.PORTUGUESE: "Olá mundo",
+        Language.ARABIC: "مرحبا بالعالم",
+        Language.HINDI: "नमस्ते दुनिया",
+    }
+    await asyncio.sleep(0.2)  # Pretend to do a network call
+    return greetings.get(to_language)
 
 
 @dataclass
@@ -33,10 +53,12 @@ def __init__(self) -> None:
             Language.ENGLISH: "Hello, world",
             Language.CHINESE: "你好，世界",
         }
+        self.lock = asyncio.Lock()
 
     @workflow.run
     async def run(self) -> str:
         await workflow.wait_condition(lambda: self.approved_for_release)
+        await workflow.wait_condition(workflow.all_handlers_finished)
         return self.greetings[self.language]
 
     @workflow.query
@@ -54,17 +76,29 @@ def approve(self, input: ApproveInput) -> None:
         self.approver_name = input.name
 
     @workflow.update
-    def set_language(self, language: Language) -> Language:
+    async def set_language(self, language: Language) -> Language:
         # 👉 An Update handler can mutate the Workflow state and return a value.
+        # 👉 Since this update handler is async, it can execute an activity.
+        if language not in self.greetings:
+            # 👉 Use a lock here to ensure that multiple calls to set_language are processed in order.
+            async with self.lock:
+                greeting = await workflow.execute_activity(
+                    call_greeting_service,
+                    language,
+                    start_to_close_timeout=timedelta(seconds=10),
+                )
+                if greeting is None:
+                    # 👉 An update validator cannot be async, so cannot be used to check that the remote
+                    # call_greeting_service supports the requested language. Raising ApplicationError
+                    # will fail the Update, but the WorkflowExecutionUpdateAccepted event will still be
+                    # added to history.
+                    raise ApplicationError(
+                        f"Greeting service does not support {language.name}"
+                    )
+                self.greetings[language] = greeting
         previous_language, self.language = self.language, language
         return previous_language
 
-    @set_language.validator
-    def validate_language(self, language: Language) -> None:
-        if language not in self.greetings:
-            # 👉 In an Update validator you raise any exception to reject the Update.
-            raise ValueError(f"{language.name} is not supported")
-
     @workflow.query
     def get_language(self) -> Language:
         return self.language