nexus sync operations (query & update) (#252)

Author: dandavison

hello_nexus/README.md

@@ -12,7 +12,7 @@ call the operations from a workflow.
 
 Start a Temporal server. (See the main samples repo [README](../README.md)).
 
-Run the following:
+Run the following to create the caller and handler namespaces, and the Nexus endpoint:
 
 ```
 temporal operator namespace create --namespace hello-nexus-basic-handler-namespace

hello_nexus/service.py

@@ -9,7 +9,8 @@
 type-safe clients, and it is used by Nexus handlers to validate that they implement
 correctly-named operation handlers with the correct input and output types.
 
-The service defined in this file features two operations: echo and hello.
+The service defined in this file exposes two operations: my_sync_operation and
+my_workflow_run_operation.
 """
 
 from dataclasses import dataclass

message_passing/introduction/starter.py

@@ -9,6 +9,7 @@
     GetLanguagesInput,
     GreetingWorkflow,
     Language,
+    SetLanguageInput,
 )
 
 
@@ -28,20 +29,20 @@ async def main(client: Optional[Client] = None):
 
     # 👉 Execute an Update
     previous_language = await wf_handle.execute_update(
-        GreetingWorkflow.set_language, Language.CHINESE
+        GreetingWorkflow.set_language, SetLanguageInput(language=Language.CHINESE)
     )
-    current_language = await wf_handle.query(GreetingWorkflow.get_language)
-    print(f"language changed: {previous_language.name} -> {current_language.name}")
+    assert await wf_handle.query(GreetingWorkflow.get_language) == Language.CHINESE
+    print(f"language changed: {previous_language.name} -> {Language.CHINESE.name}")
 
     # 👉 Start an Update and then wait for it to complete
     update_handle = await wf_handle.start_update(
         GreetingWorkflow.set_language_using_activity,
-        Language.ARABIC,
+        SetLanguageInput(language=Language.ARABIC),
         wait_for_stage=WorkflowUpdateStage.ACCEPTED,
     )
     previous_language = await update_handle.result()
-    current_language = await wf_handle.query(GreetingWorkflow.get_language)
-    print(f"language changed: {previous_language.name} -> {current_language.name}")
+    assert await wf_handle.query(GreetingWorkflow.get_language) == Language.ARABIC
+    print(f"language changed: {previous_language.name} -> {Language.ARABIC.name}")
 
     # 👉 Send a Signal
     await wf_handle.signal(GreetingWorkflow.approve, ApproveInput(name=""))

message_passing/introduction/workflows.py

@@ -16,6 +16,11 @@ class GetLanguagesInput:
     include_unsupported: bool
 
 
+@dataclass
+class SetLanguageInput:
+    language: Language
+
+
 @dataclass
 class ApproveInput:
     name: str
@@ -74,29 +79,29 @@ def approve(self, input: ApproveInput) -> None:
         self.approver_name = input.name
 
     @workflow.update
-    def set_language(self, language: Language) -> Language:
+    def set_language(self, input: SetLanguageInput) -> Language:
         # 👉 An Update handler can mutate the Workflow state and return a value.
-        previous_language, self.language = self.language, language
+        previous_language, self.language = self.language, input.language
         return previous_language
 
     @set_language.validator
-    def validate_language(self, language: Language) -> None:
-        if language not in self.greetings:
+    def validate_language(self, input: SetLanguageInput) -> None:
+        if input.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")
+            raise ValueError(f"{input.language.name} is not supported")
 
     @workflow.update
-    async def set_language_using_activity(self, language: Language) -> Language:
+    async def set_language_using_activity(self, input: SetLanguageInput) -> Language:
         # 👉 This update handler is async, so it can execute an activity.
-        if language not in self.greetings:
+        if input.language not in self.greetings:
             # 👉 We use a lock so that, if this handler is executed multiple
             # times, each execution can schedule the activity only when the
             # previously scheduled activity has completed. This ensures that
             # multiple calls to set_language are processed in order.
             async with self.lock:
                 greeting = await workflow.execute_activity(
                     call_greeting_service,
-                    language,
+                    input.language,
                     start_to_close_timeout=timedelta(seconds=10),
                 )
                 # 👉 The requested language might not be supported by the remote
@@ -108,10 +113,10 @@ async def set_language_using_activity(self, language: Language) -> Language:
                 # this purpose.)
                 if greeting is None:
                     raise ApplicationError(
-                        f"Greeting service does not support {language.name}"
+                        f"Greeting service does not support {input.language.name}"
                     )
-                self.greetings[language] = greeting
-        previous_language, self.language = self.language, language
+                self.greetings[input.language] = greeting
+        previous_language, self.language = self.language, input.language
         return previous_language
 
     @workflow.query

nexus_sync_operations/README.md

@@ -0,0 +1,39 @@
+This sample shows how to create a Nexus service that is backed by a long-running workflow and
+exposes operations that execute updates and queries against that workflow. The long-running
+workflow, and the updates/queries are private implementation detail of the nexus service: the caller
+does not know how the operations are implemented.
+
+### Sample directory structure
+
+- [service.py](./service.py) - shared Nexus service definition
+- [caller](./caller) - a caller workflow that executes Nexus operations, together with a worker and starter code
+- [handler](./handler) - Nexus operation handlers, together with a workflow used by one of the Nexus operations, and a worker that polls for both workflow, activity, and Nexus tasks.
+
+
+### Instructions
+
+Start a Temporal server. (See the main samples repo [README](../README.md)).
+
+Run the following to create the caller and handler namespaces, and the Nexus endpoint:
+
+```
+temporal operator namespace create --namespace nexus-sync-operations-handler-namespace
+temporal operator namespace create --namespace nexus-sync-operations-caller-namespace
+
+temporal operator nexus endpoint create \
+  --name nexus-sync-operations-nexus-endpoint \
+  --target-namespace nexus-sync-operations-handler-namespace \
+  --target-task-queue nexus-sync-operations-handler-task-queue \
+  --description-file nexus_sync_operations/endpoint_description.md
+```
+
+In one terminal, run the Temporal worker in the handler namespace:
+```
+uv run nexus_sync_operations/handler/worker.py
+```
+
+In another terminal, run the Temporal worker in the caller namespace and start the caller
+workflow:
+```
+uv run nexus_sync_operations/caller/app.py
+```

nexus_sync_operations/__init__.py

@@ -0,0 +1,41 @@
+import asyncio
+import uuid
+from typing import Optional
+
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from nexus_sync_operations.caller.workflows import CallerWorkflow
+
+NAMESPACE = "nexus-sync-operations-caller-namespace"
+TASK_QUEUE = "nexus-sync-operations-caller-task-queue"
+
+
+async def execute_caller_workflow(
+    client: Optional[Client] = None,
+) -> None:
+    client = client or await Client.connect(
+        "localhost:7233",
+        namespace=NAMESPACE,
+    )
+
+    async with Worker(
+        client,
+        task_queue=TASK_QUEUE,
+        workflows=[CallerWorkflow],
+    ):
+        log = await client.execute_workflow(
+            CallerWorkflow.run,
+            id=str(uuid.uuid4()),
+            task_queue=TASK_QUEUE,
+        )
+        for line in log:
+            print(line)
+
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(execute_caller_workflow())
+    except KeyboardInterrupt:
+        loop.run_until_complete(loop.shutdown_asyncgens())

nexus_sync_operations/caller/__init__.py

@@ -0,0 +1,46 @@
+"""
+This is a workflow that calls nexus operations. The caller does not have information about how these
+operations are implemented by the nexus service.
+"""
+
+from temporalio import workflow
+
+from message_passing.introduction import Language
+from message_passing.introduction.workflows import GetLanguagesInput, SetLanguageInput
+
+with workflow.unsafe.imports_passed_through():
+    from nexus_sync_operations.service import GreetingService
+
+NEXUS_ENDPOINT = "nexus-sync-operations-nexus-endpoint"
+
+
+@workflow.defn
+class CallerWorkflow:
+    @workflow.run
+    async def run(self) -> list[str]:
+        log = []
+        nexus_client = workflow.create_nexus_client(
+            service=GreetingService,
+            endpoint=NEXUS_ENDPOINT,
+        )
+
+        # Get supported languages
+        supported_languages = await nexus_client.execute_operation(
+            GreetingService.get_languages, GetLanguagesInput(include_unsupported=False)
+        )
+        log.append(f"supported languages: {supported_languages}")
+
+        # Set language
+        previous_language = await nexus_client.execute_operation(
+            GreetingService.set_language,
+            SetLanguageInput(language=Language.ARABIC),
+        )
+        assert (
+            await nexus_client.execute_operation(GreetingService.get_language, None)
+            == Language.ARABIC
+        )
+        log.append(
+            f"language changed: {previous_language.name} -> {Language.ARABIC.name}"
+        )
+
+        return log

nexus_sync_operations/caller/app.py

@@ -0,0 +1,4 @@
+## Service: [GreetingService](https://github.com/temporalio/samples-python/blob/main/nexus_sync_operations/service.py)
+- operation: `get_languages`
+- operation: `get_language`
+- operation: `set_language`