From f290a739d4eb8a5883bdee0cc7326b427fd0b2f9 Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Tue, 18 Aug 2026 16:48:06 -0500 Subject: [PATCH 1/2] Update Nexus messaging samples to use Temporal operation handlers --- nexus_messaging/callerpattern/README.md | 4 +- .../callerpattern/handler/service_handler.py | 65 ++++++++++++------- .../handler/service_handler.py | 64 +++++++++++------- uv.lock | 2 +- 4 files changed, 87 insertions(+), 48 deletions(-) diff --git a/nexus_messaging/callerpattern/README.md b/nexus_messaging/callerpattern/README.md index 9458ae43b..8be001cbc 100644 --- a/nexus_messaging/callerpattern/README.md +++ b/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: diff --git a/nexus_messaging/callerpattern/handler/service_handler.py b/nexus_messaging/callerpattern/handler/service_handler.py index cbc57eadb..8f75b5116 100644 --- a/nexus_messaging/callerpattern/handler/service_handler.py +++ b/nexus_messaging/callerpattern/handler/service_handler.py @@ -1,14 +1,15 @@ """ Nexus operation handler implementation for the entity pattern. Each operation receives a -user_id, which is mapped to a workflow ID. The operations are synchronous because queries -and updates against a running workflow complete quickly. +user_id, which is mapped to a workflow ID. 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.callerpattern.handler.workflows import GreetingWorkflow from nexus_messaging.callerpattern.service import ( @@ -38,43 +39,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()) diff --git a/nexus_messaging/ondemandpattern/handler/service_handler.py b/nexus_messaging/ondemandpattern/handler/service_handler.py index 1351aae7a..6cf12e570 100644 --- a/nexus_messaging/ondemandpattern/handler/service_handler.py +++ b/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()) diff --git a/uv.lock b/uv.lock index 202bb9448..2e07302e4 100644 --- a/uv.lock +++ b/uv.lock @@ -4655,7 +4655,7 @@ deepagents = [ { name = "langchain", marker = "python_full_version >= '3.11'", specifier = ">=1.3.11,<2" }, { name = "langchain-anthropic", marker = "python_full_version >= '3.11'", specifier = ">=1.4.7,<2" }, { name = "langchain-core", marker = "python_full_version >= '3.11'", specifier = ">=1.4.8,<2" }, - { name = "temporalio", extras = ["langsmith"], marker = "python_full_version >= '3.11'", specifier = ">=1.30.0" }, + { name = "temporalio", extras = ["langsmith"], marker = "python_full_version >= '3.11'", specifier = ">=1.31.0" }, ] dev = [ { name = "fakeredis", specifier = ">=2,<3" }, From 5efd8fe49d9611fc096b3e04a3e7dde4c42dff1c Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Tue, 18 Aug 2026 16:51:42 -0500 Subject: [PATCH 2/2] fixing comment --- nexus_messaging/callerpattern/handler/service_handler.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/nexus_messaging/callerpattern/handler/service_handler.py b/nexus_messaging/callerpattern/handler/service_handler.py index 8f75b5116..c9f384a74 100644 --- a/nexus_messaging/callerpattern/handler/service_handler.py +++ b/nexus_messaging/callerpattern/handler/service_handler.py @@ -1,8 +1,7 @@ """ Nexus operation handler implementation for the entity pattern. Each operation receives a -user_id, which is mapped to a workflow ID. Operations use Temporal operation handlers so -the SDK can manage their lifecycle and link the caller's Nexus operation to the target -Workflow. +user_id, which is mapped to a workflow ID. The operations are synchronous because queries +and updates against a running workflow complete quickly. """ from __future__ import annotations