nexus_sync_operations

dandavison · dandavison · commit 3eab7c091de6 · 2025-09-28T15:23:25.000-04:00
diff --git a/hello_nexus/README.md b/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
diff --git a/hello_nexus/service.py b/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
diff --git a/message_passing/introduction/starter.py b/message_passing/introduction/starter.py
@@ -30,8 +30,8 @@ async def main(client: Optional[Client] = None):
     previous_language = await wf_handle.execute_update(
         GreetingWorkflow.set_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(
@@ -40,8 +40,8 @@ async def main(client: Optional[Client] = None):
         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=""))
diff --git a/message_passing/introduction/workflows.py b/message_passing/introduction/workflows.py
@@ -16,6 +16,16 @@ class GetLanguagesInput:
     include_unsupported: bool
 
 
+@dataclass
+class SetLanguageInput:
+    language: Language
+
+
+@dataclass
+class SetLanguageUsingActivityInput:
+    language: Language
+
+
 @dataclass
 class ApproveInput:
     name: str
diff --git a/nexus_sync_operations/README.md b/nexus_sync_operations/README.md
@@ -0,0 +1,37 @@
+This sample shows how to create a Nexus service that is backed by a long-running workflow and
+exposes operations that use signals, queries, and updates against that workflow.
+
+### 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
+```
diff --git a/nexus_sync_operations/__init__.py b/nexus_sync_operations/__init__.py
diff --git a/nexus_sync_operations/caller/__init__.py b/nexus_sync_operations/caller/__init__.py
diff --git a/nexus_sync_operations/caller/app.py b/nexus_sync_operations/caller/app.py
@@ -0,0 +1,39 @@
+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],
+    ):
+        await client.execute_workflow(
+            CallerWorkflow.run,
+            id=str(uuid.uuid4()),
+            task_queue=TASK_QUEUE,
+        )
+
+
+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())
diff --git a/nexus_sync_operations/caller/workflows.py b/nexus_sync_operations/caller/workflows.py
@@ -0,0 +1,55 @@
+from message_passing.introduction import Language
+from message_passing.introduction.workflows import (
+    ApproveInput,
+    GetLanguagesInput,
+    SetLanguageInput,
+    SetLanguageUsingActivityInput,
+)
+from temporalio import workflow
+
+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) -> None:
+        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)
+        )
+        print(f"supported languages: {supported_languages}")
+
+        # Set language
+        previous_language = await nexus_client.execute_operation(
+            GreetingService.set_language, SetLanguageInput(language=Language.CHINESE)
+        )
+        assert (
+            await nexus_client.execute_operation(GreetingService.get_language, None)
+            == Language.CHINESE
+        )
+        print(f"language changed: {previous_language.name} -> {Language.CHINESE.name}")
+
+        # Set language using remote service
+        previous_language = await nexus_client.execute_operation(
+            GreetingService.set_language_using_activity,
+            SetLanguageUsingActivityInput(language=Language.ARABIC),
+        )
+        assert (
+            await nexus_client.execute_operation(GreetingService.get_language, None)
+            == Language.ARABIC
+        )
+        print(f"language changed: {previous_language.name} -> {Language.ARABIC.name}")
+
+        # Approve
+        await nexus_client.execute_operation(
+            GreetingService.approve, ApproveInput(name="")
+        )
diff --git a/nexus_sync_operations/endpoint_description.md b/nexus_sync_operations/endpoint_description.md
@@ -0,0 +1,6 @@
+## Service: [GreetingService](https://github.com/temporalio/samples-python/blob/main/nexus_sync_operations/service.py)
+- operation: `get_languages`
+- operation: `get_language`
+- operation: `set_language`
+- operation: `set_language_using_activity`
+- operation: `approve`
diff --git a/nexus_sync_operations/handler/__init__.py b/nexus_sync_operations/handler/__init__.py
diff --git a/nexus_sync_operations/handler/service_handler.py b/nexus_sync_operations/handler/service_handler.py
@@ -0,0 +1,69 @@
+"""
+This file demonstrates how to implement a Nexus service that is backed by a long-running
+workflow and exposes operations that perform signals, updates, and queries against that
+workflow.
+"""
+
+from __future__ import annotations
+
+import nexusrpc
+from temporalio.client import WorkflowHandle
+
+from message_passing.introduction import Language
+from message_passing.introduction.workflows import (
+    ApproveInput,
+    GetLanguagesInput,
+    GreetingWorkflow,
+    SetLanguageInput,
+    SetLanguageUsingActivityInput,
+)
+from nexus_sync_operations.service import GreetingService
+
+
+@nexusrpc.handler.service_handler(service=GreetingService)
+class GreetingServiceHandler:
+    def __init__(
+        self,
+        greeting_workflow_handle: WorkflowHandle[GreetingWorkflow, str],
+    ):
+        self.greeting_workflow_handle = greeting_workflow_handle
+
+    @nexusrpc.handler.sync_operation
+    async def get_languages(
+        self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguagesInput
+    ) -> list[Language]:
+        return await self.greeting_workflow_handle.query(
+            GreetingWorkflow.get_languages, input
+        )
+
+    @nexusrpc.handler.sync_operation
+    async def get_language(
+        self, ctx: nexusrpc.handler.StartOperationContext, input: None
+    ) -> Language:
+        return await self.greeting_workflow_handle.query(GreetingWorkflow.get_language)
+
+    @nexusrpc.handler.sync_operation
+    async def set_language(
+        self, ctx: nexusrpc.handler.StartOperationContext, input: SetLanguageInput
+    ) -> Language:
+        return await self.greeting_workflow_handle.execute_update(
+            GreetingWorkflow.set_language, input.language
+        )
+
+    @nexusrpc.handler.sync_operation
+    async def set_language_using_activity(
+        self,
+        ctx: nexusrpc.handler.StartOperationContext,
+        input: SetLanguageUsingActivityInput,
+    ) -> Language:
+        return await self.greeting_workflow_handle.execute_update(
+            GreetingWorkflow.set_language_using_activity, input.language
+        )
+
+    @nexusrpc.handler.sync_operation
+    async def approve(
+        self, ctx: nexusrpc.handler.StartOperationContext, input: ApproveInput
+    ) -> None:
+        return await self.greeting_workflow_handle.signal(
+            GreetingWorkflow.approve, input
+        )
diff --git a/nexus_sync_operations/handler/worker.py b/nexus_sync_operations/handler/worker.py
@@ -0,0 +1,57 @@
+import asyncio
+import logging
+from typing import Optional
+
+from temporalio.client import Client
+from temporalio.common import WorkflowIDConflictPolicy
+from temporalio.worker import Worker
+
+from message_passing.introduction.activities import call_greeting_service
+from message_passing.introduction.workflows import GreetingWorkflow
+from nexus_sync_operations.handler.service_handler import GreetingServiceHandler
+
+interrupt_event = asyncio.Event()
+
+NAMESPACE = "nexus-sync-operations-handler-namespace"
+TASK_QUEUE = "nexus-sync-operations-handler-task-queue"
+
+
+async def main(client: Optional[Client] = None):
+    logging.basicConfig(level=logging.INFO)
+
+    client = client or await Client.connect(
+        "localhost:7233",
+        namespace=NAMESPACE,
+    )
+
+    async with Worker(
+        client,
+        task_queue=TASK_QUEUE,
+        workflows=[GreetingWorkflow],
+        activities=[call_greeting_service],
+    ):
+        long_running_workflow_handle = await client.start_workflow(
+            GreetingWorkflow.run,
+            id="nexus-sync-operations-greeting-workflow",
+            task_queue=TASK_QUEUE,
+            id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
+        )
+        async with Worker(
+            client,
+            task_queue=TASK_QUEUE,
+            nexus_service_handlers=[
+                GreetingServiceHandler(long_running_workflow_handle)
+            ],
+        ):
+            logging.info("Worker started, ctrl+c to exit")
+            await interrupt_event.wait()
+            logging.info("Shutting down")
+
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        loop.run_until_complete(loop.shutdown_asyncgens())
diff --git a/nexus_sync_operations/service.py b/nexus_sync_operations/service.py
@@ -0,0 +1,19 @@
+import nexusrpc
+from message_passing.introduction import Language
+from message_passing.introduction.workflows import (
+    ApproveInput,
+    GetLanguagesInput,
+    SetLanguageInput,
+    SetLanguageUsingActivityInput,
+)
+
+
+@nexusrpc.service
+class GreetingService:
+    get_languages: nexusrpc.Operation[GetLanguagesInput, list[Language]]
+    get_language: nexusrpc.Operation[None, Language]
+    set_language: nexusrpc.Operation[SetLanguageInput, Language]
+    set_language_using_activity: nexusrpc.Operation[
+        SetLanguageUsingActivityInput, Language
+    ]
+    approve: nexusrpc.Operation[ApproveInput, None]
diff --git a/tests/nexus_sync_operations/nexus_sync_operations_test.py b/tests/nexus_sync_operations/nexus_sync_operations_test.py
