Nexus query sdk ergonomics

nexus_messaging/callerpattern/README.md

@@ -1,12 +1,12 @@
 ## Caller pattern
 
 The handler worker starts a `GreetingWorkflow` for a User ID.
-`NexusGreetingServiceHandler` holds that ID and routes every Nexus operation to it.
+`NexusGreetingServiceHandler` derives the Workflow ID and routes every Nexus operation to it.
 The caller's input does not have that Workflow ID as the caller doesn't know it -- but the caller
 sends in the User ID, and `NexusGreetingServiceHandler` knows how to get the desired Workflow ID
 from that User ID (see the `get_workflow_id` call).
 
-The handler worker uses the same `get_workflow_id` call to generate a Workflow ID from a Wser ID
+The handler worker uses the same `get_workflow_id` call to generate a Workflow ID from a User ID
 when it launches the Workflow.
 
 The caller Workflow:

nexus_messaging/callerpattern/handler/service_handler.py

@@ -8,7 +8,7 @@
 
 import nexusrpc
 from temporalio import nexus
-from temporalio.client import WorkflowHandle
+from temporalio.client import Client, WorkflowHandle
 
 from nexus_messaging.callerpattern.handler.workflows import GreetingWorkflow
 from nexus_messaging.callerpattern.service import (
@@ -38,43 +38,61 @@ def get_workflow_id(user_id: str) -> str:
 @nexusrpc.handler.service_handler(service=NexusGreetingService)
 class NexusGreetingServiceHandler:
     def _get_workflow_handle(
-        self, user_id: str
+        self, client: Client, user_id: str
     ) -> WorkflowHandle[GreetingWorkflow, str]:
-        return nexus.client().get_workflow_handle_for(
+        return client.get_workflow_handle_for(
             GreetingWorkflow.run, get_workflow_id(user_id)
         )
 
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def get_languages(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguagesInput
-    ) -> GetLanguagesOutput:
-        return await self._get_workflow_handle(input.user_id).query(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: GetLanguagesInput,
+    ) -> nexus.TemporalOperationResult[GetLanguagesOutput]:
+        result = await self._get_workflow_handle(client.client, input.user_id).query(
             GreetingWorkflow.get_languages, input
         )
+        return nexus.TemporalOperationResult.sync(result)
 
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def get_language(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguageInput
-    ) -> Language:
-        return await self._get_workflow_handle(input.user_id).query(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: GetLanguageInput,
+    ) -> nexus.TemporalOperationResult[Language]:
+        result = await self._get_workflow_handle(client.client, input.user_id).query(
             GreetingWorkflow.get_language
         )
+        return nexus.TemporalOperationResult.sync(result)
 
     # Routes to set_language_using_activity (not set_language) so that new languages not
     # already in the greetings map can be fetched via an activity.
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def set_language(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: SetLanguageInput
-    ) -> Language:
-        return await self._get_workflow_handle(input.user_id).execute_update(
-            GreetingWorkflow.set_language_using_activity, input
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: SetLanguageInput,
+    ) -> nexus.TemporalOperationResult[Language]:
+        result = await self._get_workflow_handle(
+            client.client, input.user_id
+        ).execute_update(
+            GreetingWorkflow.set_language_using_activity,
+            input,
         )
+        return nexus.TemporalOperationResult.sync(result)
 
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def approve(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: ApproveInput
-    ) -> ApproveOutput:
-        await self._get_workflow_handle(input.user_id).signal(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: ApproveInput,
+    ) -> nexus.TemporalOperationResult[ApproveOutput]:
+        await self._get_workflow_handle(client.client, input.user_id).signal(
             GreetingWorkflow.approve, input
         )
-        return ApproveOutput()
+        return nexus.TemporalOperationResult.sync(ApproveOutput())

nexus_messaging/ondemandpattern/handler/service_handler.py

@@ -1,13 +1,15 @@
 """
 Nexus operation handler for the on-demand pattern. Each operation receives the target
-userId in its input, and run_from_remote starts a brand-new GreetingWorkflow.
+user_id in its input, and run_from_remote starts a brand-new GreetingWorkflow. Operations
+use Temporal operation handlers so the SDK can manage their lifecycle and link the caller's
+Nexus operation to the target Workflow.
 """
 
 from __future__ import annotations
 
 import nexusrpc
 from temporalio import nexus
-from temporalio.client import WorkflowHandle
+from temporalio.client import Client, WorkflowHandle
 
 from nexus_messaging.ondemandpattern.handler.workflows import GreetingWorkflow
 from nexus_messaging.ondemandpattern.service import (
@@ -31,9 +33,9 @@ def _get_workflow_id(self, user_id: str) -> str:
         return WORKFLOW_ID_PREFIX + user_id
 
     def _get_workflow_handle(
-        self, user_id: str
+        self, client: Client, user_id: str
     ) -> WorkflowHandle[GreetingWorkflow, str]:
-        return nexus.client().get_workflow_handle_for(
+        return client.get_workflow_handle_for(
             GreetingWorkflow.run, self._get_workflow_id(user_id)
         )
 
@@ -48,37 +50,55 @@ async def run_from_remote(
             id=self._get_workflow_id(input.user_id),
         )
 
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def get_languages(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguagesInput
-    ) -> GetLanguagesOutput:
-        return await self._get_workflow_handle(input.user_id).query(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: GetLanguagesInput,
+    ) -> nexus.TemporalOperationResult[GetLanguagesOutput]:
+        result = await self._get_workflow_handle(client.client, input.user_id).query(
             GreetingWorkflow.get_languages, input
         )
+        return nexus.TemporalOperationResult.sync(result)
 
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def get_language(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguageInput
-    ) -> Language:
-        return await self._get_workflow_handle(input.user_id).query(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: GetLanguageInput,
+    ) -> nexus.TemporalOperationResult[Language]:
+        result = await self._get_workflow_handle(client.client, input.user_id).query(
             GreetingWorkflow.get_language,
         )
+        return nexus.TemporalOperationResult.sync(result)
 
     # Routes to set_language_using_activity so that new languages not already in the
     # greetings map can be fetched via an activity.
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def set_language(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: SetLanguageInput
-    ) -> Language:
-        return await self._get_workflow_handle(input.user_id).execute_update(
-            GreetingWorkflow.set_language_using_activity, input
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: SetLanguageInput,
+    ) -> nexus.TemporalOperationResult[Language]:
+        result = await self._get_workflow_handle(
+            client.client, input.user_id
+        ).execute_update(
+            GreetingWorkflow.set_language_using_activity,
+            input,
         )
+        return nexus.TemporalOperationResult.sync(result)
 
-    @nexusrpc.handler.sync_operation
+    @nexus.temporal_operation
     async def approve(
-        self, ctx: nexusrpc.handler.StartOperationContext, input: ApproveInput
-    ) -> ApproveOutput:
-        await self._get_workflow_handle(input.user_id).signal(
+        self,
+        _ctx: nexus.TemporalStartOperationContext,
+        client: nexus.TemporalNexusClient,
+        input: ApproveInput,
+    ) -> nexus.TemporalOperationResult[ApproveOutput]:
+        await self._get_workflow_handle(client.client, input.user_id).signal(
             GreetingWorkflow.approve, input
         )
-        return ApproveOutput()
+        return nexus.TemporalOperationResult.sync(ApproveOutput())