Cleanup

Author: dandavison

nexus_sync_operations/README.md

@@ -1,5 +1,7 @@
 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.
+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
 

nexus_sync_operations/caller/workflows.py

@@ -1,10 +1,12 @@
+"""
+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,
-)
+from message_passing.introduction.workflows import GetLanguagesInput, SetLanguageInput
 
 with workflow.unsafe.imports_passed_through():
     from nexus_sync_operations.service import GreetingService

nexus_sync_operations/endpoint_description.md

@@ -2,5 +2,3 @@
 - operation: `get_languages`
 - operation: `get_language`
 - operation: `set_language`
-- operation: `set_language_using_activity`
-- operation: `approve`

nexus_sync_operations/handler/service_handler.py

@@ -1,12 +1,12 @@
 """
-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.
+This file demonstrates how to implement a Nexus service that is backed by a long-running workflow
+and exposes operations that perform updates, and queries against that workflow.
 """
 
 from __future__ import annotations
 
 import nexusrpc
+from temporalio import nexus
 from temporalio.client import Client, WorkflowHandle
 from temporalio.common import WorkflowIDConflictPolicy
 
@@ -21,29 +21,38 @@
 
 @nexusrpc.handler.service_handler(service=GreetingService)
 class GreetingServiceHandler:
-    def __init__(
-        self,
-        greeting_workflow_handle: WorkflowHandle[GreetingWorkflow, str],
-    ):
-        self.greeting_workflow_handle = greeting_workflow_handle
+    # This nexus service is backed by a long-running "entity" workflow. This means that the workflow
+    # is always running in the background, allowing the service to be stateful and durable. The
+    # service interacts with it via messages (updates and queries). All of this is implementation
+    # detail private to the nexus handler: the nexus caller does not know how the operations are
+    # implemented or what is providing the backing storage.
+    LONG_RUNNING_WORKFLOW_ID = "nexus-sync-operations-greeting-workflow"
 
     @classmethod
-    async def create(cls, client: Client, task_queue: str) -> GreetingServiceHandler:
-        # Obtain a workflow handle to the long-running workflow that backs this service, starting
-        # the workflow if it is not already running.
-        return cls(await cls._get_workflow_handle(client, task_queue))
-
-    @staticmethod
-    async def _get_workflow_handle(
-        client: Client, task_queue: str
-    ) -> WorkflowHandle[GreetingWorkflow, str]:
-        return await client.start_workflow(
+    async def start(cls, client: Client, task_queue: str) -> None:
+        # Start the long-running "entity" workflow, if it is not already running.
+        await client.start_workflow(
             GreetingWorkflow.run,
-            id="nexus-sync-operations-greeting-workflow",
+            id=cls.LONG_RUNNING_WORKFLOW_ID,
             task_queue=task_queue,
             id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
         )
 
+    @property
+    def greeting_workflow_handle(self) -> WorkflowHandle[GreetingWorkflow, str]:
+        # In nexus operation handler code, nexus.client() is always available, returning a client
+        # connected to the handler namespace (it's the same client instance that your nexus worker
+        # is using to poll the server for nexus tasks). This client can be used to interact with the
+        # handler namespace, for example to send signals, queries, or updates. Remember however,
+        # that a sync_operation handler must return quickly (no more than a few seconds). To do
+        # long-running work in a nexus operation handler, use
+        # temporalio.nexus.workflow_run_operation (see the hello_nexus sample).
+        return nexus.client().get_workflow_handle_for(
+            GreetingWorkflow.run, self.LONG_RUNNING_WORKFLOW_ID
+        )
+
+    # 👉 This is a handler for a nexus operation whose internal implementation involves executing a
+    # query against a long-running workflow that is private to the nexus service.
     @nexusrpc.handler.sync_operation
     async def get_languages(
         self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguagesInput
@@ -52,12 +61,16 @@ async def get_languages(
             GreetingWorkflow.get_languages, input
         )
 
+    # 👉 This is a handler for a nexus operation whose internal implementation involves executing a
+    # query against a long-running workflow that is private to the nexus service.
     @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)
 
+    # 👉 This is a handler for a nexus operation whose internal implementation involves executing an
+    # update against a long-running workflow that is private to the nexus service.
     @nexusrpc.handler.sync_operation
     async def set_language(
         self,

nexus_sync_operations/handler/worker.py

@@ -22,7 +22,8 @@ async def main(client: Optional[Client] = None):
         "localhost:7233",
         namespace=NAMESPACE,
     )
-    greeting_service_handler = await GreetingServiceHandler.create(client, TASK_QUEUE)
+    greeting_service_handler = GreetingServiceHandler()
+    await greeting_service_handler.start(client, TASK_QUEUE)
 
     async with Worker(
         client,

nexus_sync_operations/service.py

@@ -1,10 +1,16 @@
+"""
+This module defines a Nexus service that exposes three operations.
+
+It is used by the nexus service handler to validate that the operation handlers implement the
+correct input and output types, and by the caller workflow to create a type-safe client. It does not
+contain the implementation of the operations; see nexus_sync_operations.handler.service_handler for
+that.
+"""
+
 import nexusrpc
 
 from message_passing.introduction import Language
-from message_passing.introduction.workflows import (
-    GetLanguagesInput,
-    SetLanguageInput,
-)
+from message_passing.introduction.workflows import GetLanguagesInput, SetLanguageInput
 
 
 @nexusrpc.service