Skip to content

Commit 7174ed7

Browse files
JoshuaFrenchwoodJoshuaFrenchwood
authored
Nexus query sdk ergonomics (#352)
* Update Nexus messaging samples to use Temporal operation handlers * fixing comment
1 parent 05070f6 commit 7174ed7

4 files changed

Lines changed: 84 additions & 46 deletions

File tree

‎nexus_messaging/callerpattern/README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
## Caller pattern
22

33
The handler worker starts a `GreetingWorkflow` for a User ID.
4-
`NexusGreetingServiceHandler` holds that ID and routes every Nexus operation to it.
4+
`NexusGreetingServiceHandler` derives the Workflow ID and routes every Nexus operation to it.
55
The caller's input does not have that Workflow ID as the caller doesn't know it -- but the caller
66
sends in the User ID, and `NexusGreetingServiceHandler` knows how to get the desired Workflow ID
77
from that User ID (see the `get_workflow_id` call).
88

9-
The handler worker uses the same `get_workflow_id` call to generate a Workflow ID from a Wser ID
9+
The handler worker uses the same `get_workflow_id` call to generate a Workflow ID from a User ID
1010
when it launches the Workflow.
1111

1212
The caller Workflow:

‎nexus_messaging/callerpattern/handler/service_handler.py‎

Lines changed: 39 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
import nexusrpc
1010
from temporalio import nexus
11-
from temporalio.client import WorkflowHandle
11+
from temporalio.client import Client, WorkflowHandle
1212

1313
from nexus_messaging.callerpattern.handler.workflows import GreetingWorkflow
1414
from nexus_messaging.callerpattern.service import (
@@ -38,43 +38,61 @@ def get_workflow_id(user_id: str) -> str:
3838
@nexusrpc.handler.service_handler(service=NexusGreetingService)
3939
class NexusGreetingServiceHandler:
4040
def _get_workflow_handle(
41-
self, user_id: str
41+
self, client: Client, user_id: str
4242
) -> WorkflowHandle[GreetingWorkflow, str]:
43-
return nexus.client().get_workflow_handle_for(
43+
return client.get_workflow_handle_for(
4444
GreetingWorkflow.run, get_workflow_id(user_id)
4545
)
4646

47-
@nexusrpc.handler.sync_operation
47+
@nexus.temporal_operation
4848
async def get_languages(
49-
self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguagesInput
50-
) -> GetLanguagesOutput:
51-
return await self._get_workflow_handle(input.user_id).query(
49+
self,
50+
_ctx: nexus.TemporalStartOperationContext,
51+
client: nexus.TemporalNexusClient,
52+
input: GetLanguagesInput,
53+
) -> nexus.TemporalOperationResult[GetLanguagesOutput]:
54+
result = await self._get_workflow_handle(client.client, input.user_id).query(
5255
GreetingWorkflow.get_languages, input
5356
)
57+
return nexus.TemporalOperationResult.sync(result)
5458

55-
@nexusrpc.handler.sync_operation
59+
@nexus.temporal_operation
5660
async def get_language(
57-
self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguageInput
58-
) -> Language:
59-
return await self._get_workflow_handle(input.user_id).query(
61+
self,
62+
_ctx: nexus.TemporalStartOperationContext,
63+
client: nexus.TemporalNexusClient,
64+
input: GetLanguageInput,
65+
) -> nexus.TemporalOperationResult[Language]:
66+
result = await self._get_workflow_handle(client.client, input.user_id).query(
6067
GreetingWorkflow.get_language
6168
)
69+
return nexus.TemporalOperationResult.sync(result)
6270

6371
# Routes to set_language_using_activity (not set_language) so that new languages not
6472
# already in the greetings map can be fetched via an activity.
65-
@nexusrpc.handler.sync_operation
73+
@nexus.temporal_operation
6674
async def set_language(
67-
self, ctx: nexusrpc.handler.StartOperationContext, input: SetLanguageInput
68-
) -> Language:
69-
return await self._get_workflow_handle(input.user_id).execute_update(
70-
GreetingWorkflow.set_language_using_activity, input
75+
self,
76+
_ctx: nexus.TemporalStartOperationContext,
77+
client: nexus.TemporalNexusClient,
78+
input: SetLanguageInput,
79+
) -> nexus.TemporalOperationResult[Language]:
80+
result = await self._get_workflow_handle(
81+
client.client, input.user_id
82+
).execute_update(
83+
GreetingWorkflow.set_language_using_activity,
84+
input,
7185
)
86+
return nexus.TemporalOperationResult.sync(result)
7287

73-
@nexusrpc.handler.sync_operation
88+
@nexus.temporal_operation
7489
async def approve(
75-
self, ctx: nexusrpc.handler.StartOperationContext, input: ApproveInput
76-
) -> ApproveOutput:
77-
await self._get_workflow_handle(input.user_id).signal(
90+
self,
91+
_ctx: nexus.TemporalStartOperationContext,
92+
client: nexus.TemporalNexusClient,
93+
input: ApproveInput,
94+
) -> nexus.TemporalOperationResult[ApproveOutput]:
95+
await self._get_workflow_handle(client.client, input.user_id).signal(
7896
GreetingWorkflow.approve, input
7997
)
80-
return ApproveOutput()
98+
return nexus.TemporalOperationResult.sync(ApproveOutput())

‎nexus_messaging/ondemandpattern/handler/service_handler.py‎

Lines changed: 42 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,15 @@
11
"""
22
Nexus operation handler for the on-demand pattern. Each operation receives the target
3-
userId in its input, and run_from_remote starts a brand-new GreetingWorkflow.
3+
user_id in its input, and run_from_remote starts a brand-new GreetingWorkflow. Operations
4+
use Temporal operation handlers so the SDK can manage their lifecycle and link the caller's
5+
Nexus operation to the target Workflow.
46
"""
57

68
from __future__ import annotations
79

810
import nexusrpc
911
from temporalio import nexus
10-
from temporalio.client import WorkflowHandle
12+
from temporalio.client import Client, WorkflowHandle
1113

1214
from nexus_messaging.ondemandpattern.handler.workflows import GreetingWorkflow
1315
from nexus_messaging.ondemandpattern.service import (
@@ -31,9 +33,9 @@ def _get_workflow_id(self, user_id: str) -> str:
3133
return WORKFLOW_ID_PREFIX + user_id
3234

3335
def _get_workflow_handle(
34-
self, user_id: str
36+
self, client: Client, user_id: str
3537
) -> WorkflowHandle[GreetingWorkflow, str]:
36-
return nexus.client().get_workflow_handle_for(
38+
return client.get_workflow_handle_for(
3739
GreetingWorkflow.run, self._get_workflow_id(user_id)
3840
)
3941

@@ -48,37 +50,55 @@ async def run_from_remote(
4850
id=self._get_workflow_id(input.user_id),
4951
)
5052

51-
@nexusrpc.handler.sync_operation
53+
@nexus.temporal_operation
5254
async def get_languages(
53-
self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguagesInput
54-
) -> GetLanguagesOutput:
55-
return await self._get_workflow_handle(input.user_id).query(
55+
self,
56+
_ctx: nexus.TemporalStartOperationContext,
57+
client: nexus.TemporalNexusClient,
58+
input: GetLanguagesInput,
59+
) -> nexus.TemporalOperationResult[GetLanguagesOutput]:
60+
result = await self._get_workflow_handle(client.client, input.user_id).query(
5661
GreetingWorkflow.get_languages, input
5762
)
63+
return nexus.TemporalOperationResult.sync(result)
5864

59-
@nexusrpc.handler.sync_operation
65+
@nexus.temporal_operation
6066
async def get_language(
61-
self, ctx: nexusrpc.handler.StartOperationContext, input: GetLanguageInput
62-
) -> Language:
63-
return await self._get_workflow_handle(input.user_id).query(
67+
self,
68+
_ctx: nexus.TemporalStartOperationContext,
69+
client: nexus.TemporalNexusClient,
70+
input: GetLanguageInput,
71+
) -> nexus.TemporalOperationResult[Language]:
72+
result = await self._get_workflow_handle(client.client, input.user_id).query(
6473
GreetingWorkflow.get_language,
6574
)
75+
return nexus.TemporalOperationResult.sync(result)
6676

6777
# Routes to set_language_using_activity so that new languages not already in the
6878
# greetings map can be fetched via an activity.
69-
@nexusrpc.handler.sync_operation
79+
@nexus.temporal_operation
7080
async def set_language(
71-
self, ctx: nexusrpc.handler.StartOperationContext, input: SetLanguageInput
72-
) -> Language:
73-
return await self._get_workflow_handle(input.user_id).execute_update(
74-
GreetingWorkflow.set_language_using_activity, input
81+
self,
82+
_ctx: nexus.TemporalStartOperationContext,
83+
client: nexus.TemporalNexusClient,
84+
input: SetLanguageInput,
85+
) -> nexus.TemporalOperationResult[Language]:
86+
result = await self._get_workflow_handle(
87+
client.client, input.user_id
88+
).execute_update(
89+
GreetingWorkflow.set_language_using_activity,
90+
input,
7591
)
92+
return nexus.TemporalOperationResult.sync(result)
7693

77-
@nexusrpc.handler.sync_operation
94+
@nexus.temporal_operation
7895
async def approve(
79-
self, ctx: nexusrpc.handler.StartOperationContext, input: ApproveInput
80-
) -> ApproveOutput:
81-
await self._get_workflow_handle(input.user_id).signal(
96+
self,
97+
_ctx: nexus.TemporalStartOperationContext,
98+
client: nexus.TemporalNexusClient,
99+
input: ApproveInput,
100+
) -> nexus.TemporalOperationResult[ApproveOutput]:
101+
await self._get_workflow_handle(client.client, input.user_id).signal(
82102
GreetingWorkflow.approve, input
83103
)
84-
return ApproveOutput()
104+
return nexus.TemporalOperationResult.sync(ApproveOutput())

‎uv.lock‎

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)