Skip to content

Commit 2affa25

Browse files
authored
Python Nexus convert workflow start errors to Nexus errors (#1052)
1 parent a9c71aa commit 2affa25

File tree

2 files changed

+32
-2
lines changed

2 files changed

+32
-2
lines changed

temporalio/worker/_nexus.py

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@
3232
import temporalio.common
3333
import temporalio.converter
3434
import temporalio.nexus
35-
from temporalio.exceptions import ApplicationError
35+
from temporalio.exceptions import ApplicationError, WorkflowAlreadyStartedError
3636
from temporalio.nexus import Info, logger
3737
from temporalio.service import RPCError, RPCStatusCode
3838

@@ -445,6 +445,12 @@ def _exception_to_handler_error(err: BaseException) -> nexusrpc.HandlerError:
445445
type=nexusrpc.HandlerErrorType.INTERNAL,
446446
retryable_override=not err.non_retryable,
447447
)
448+
elif isinstance(err, WorkflowAlreadyStartedError):
449+
handler_err = nexusrpc.HandlerError(
450+
err.message,
451+
type=nexusrpc.HandlerErrorType.INTERNAL,
452+
retryable_override=False,
453+
)
448454
elif isinstance(err, RPCError):
449455
if err.status == RPCStatusCode.INVALID_ARGUMENT:
450456
handler_err = nexusrpc.HandlerError(

tests/nexus/test_workflow_caller_errors.py

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
sync_operation,
2222
)
2323

24-
from temporalio import workflow
24+
from temporalio import nexus, workflow
2525
from temporalio.client import (
2626
Client,
2727
WorkflowFailureError,
@@ -47,6 +47,13 @@ class ErrorTestInput:
4747
id: str
4848

4949

50+
@workflow.defn
51+
class NonTerminatingWorkflow:
52+
@workflow.run
53+
async def run(self) -> None:
54+
await asyncio.Event().wait()
55+
56+
5057
@nexusrpc.handler.service_handler
5158
class ErrorTestService:
5259
@nexusrpc.handler.sync_operation
@@ -87,6 +94,18 @@ def retried_due_to_internal_handler_error(
8794
type=nexusrpc.HandlerErrorType.INTERNAL,
8895
)
8996

97+
@nexusrpc.handler.sync_operation
98+
async def fails_due_to_workflow_already_started(
99+
self, ctx: nexusrpc.handler.StartOperationContext, input: ErrorTestInput
100+
) -> None:
101+
operation_invocation_counts[input.id] += 1
102+
for _ in range(2):
103+
await nexus.client().start_workflow(
104+
NonTerminatingWorkflow.run,
105+
id="second-start-request-will-fail",
106+
task_queue=nexus.info().task_queue,
107+
)
108+
90109

91110
@workflow.defn(sandboxed=False)
92111
class CallerWorkflow:
@@ -156,6 +175,11 @@ async def times_called() -> int:
156175
nexusrpc.HandlerErrorType.NOT_FOUND,
157176
"No handler for service",
158177
),
178+
(
179+
"fails_due_to_workflow_already_started",
180+
nexusrpc.HandlerErrorType.INTERNAL,
181+
"already started",
182+
),
159183
],
160184
)
161185
async def test_nexus_operation_fails_without_retry_as_handler_error(

0 commit comments

Comments
 (0)