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..c9f384a74 100644 --- a/nexus_messaging/callerpattern/handler/service_handler.py +++ b/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()) 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" },