diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 613f923b2e..c7a0065a77 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -229,7 +229,7 @@ jobs: - run: uv add --dev --python 3.10 "googleapis-common-protos==1.70.0" - run: uv add --python 3.10 "protobuf<4" - run: uv sync --all-extras - - run: cargo install --locked nexgen --version 0.2.8 --features advanced --force + - run: cargo install --locked nexgen --version 0.2.11 --features advanced --force - run: poe build-develop - run: poe gen-protos - name: Check generation unchanged diff --git a/changelog/fixed/wandering-walrus.md b/changelog/fixed/wandering-walrus.md new file mode 100644 index 0000000000..973f00c09c --- /dev/null +++ b/changelog/fixed/wandering-walrus.md @@ -0,0 +1 @@ +System Nexus operations with no generated interceptor now go to the generic `start_system_nexus_operation` hook of the Workflow outbound interceptors, instead of raising. diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index be72cf37a9..c37df01812 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -25,7 +25,7 @@ workflow_output_dir = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" notification_output_dir = base_dir / "temporalio" / "nexus" / "notifications" workflow_init_path = base_dir / "temporalio" / "workflow" / "__init__.py" -NEX_GEN_VERSION = "0.2.8" +NEX_GEN_VERSION = "0.2.11" proto_files = [ gen_protos.api_proto_dir / "temporal" diff --git a/temporalio/nexus/notifications/__init__.py b/temporalio/nexus/notifications/__init__.py index 7a8bb24ead..6b8a5bdbba 100644 --- a/temporalio/nexus/notifications/__init__.py +++ b/temporalio/nexus/notifications/__init__.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/notifications/models.py b/temporalio/nexus/notifications/models.py index 64ad07fbae..ef5528a6f8 100644 --- a/temporalio/nexus/notifications/models.py +++ b/temporalio/nexus/notifications/models.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index f6a6f0c682..1d3efca427 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py index 5c62e00927..c1e992c8c7 100644 --- a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py +++ b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations @@ -40,9 +40,7 @@ async def _start_system_nexus_operation( "temporalio.workflow.NexusOperationHandle[_OutputT]", await interceptor.start_signal_with_start_workflow(typed_input.input), ) - raise ValueError( - f"unsupported System Nexus operation: {input.service}/{input.operation_name}" - ) + return await interceptor.start_system_nexus_operation(input) class _SystemNexusWorkflowOutboundInterceptorBase(abc.ABC): @@ -51,6 +49,14 @@ def _next_system_nexus_interceptor( self, ) -> _SystemNexusWorkflowOutboundInterceptorBase: ... + @abc.abstractmethod + async def start_system_nexus_operation( + self, + input: StartNexusOperationInput[_InputT, _OutputT], + ) -> temporalio.workflow.NexusOperationHandle[_OutputT]: + """Intercept a System Nexus operation through the generic hook.""" + ... + async def start_signal_with_start_workflow( self, request: models.SignalWithStartWorkflowRequest ) -> temporalio.workflow.NexusOperationHandle[ diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index dc7b5ce2e7..8fa35902fc 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations @@ -18,7 +18,8 @@ @dataclasses.dataclass(slots=True, kw_only=True) class SignalWithStartWorkflowRequest: - """ + """Request fields for signaling a workflow, starting it first if needed. + .. warning:: This API is experimental and subject to change. """ @@ -223,7 +224,8 @@ def to_transfer_type( @dataclasses.dataclass(slots=True) class UserMetadata: - """ + """Static metadata for a workflow execution. + .. warning:: This API is experimental and subject to change. """ @@ -276,7 +278,8 @@ def to_transfer_type( @dataclasses.dataclass(slots=True) class SignalWithStartWorkflowResponse: - """ + """Result of signaling a workflow and starting it if needed. + .. warning:: This API is experimental and subject to change. """ diff --git a/temporalio/nexus/system/workflow_service/operations/__init__.py b/temporalio/nexus/system/workflow_service/operations/__init__.py index ff561b55c0..3bd09999cf 100644 --- a/temporalio/nexus/system/workflow_service/operations/__init__.py +++ b/temporalio/nexus/system/workflow_service/operations/__init__.py @@ -1,3 +1,3 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py index 28b4735436..057c6078ed 100644 --- a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py +++ b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/services.py b/temporalio/nexus/system/workflow_service/services.py index c95fff35ad..61cf577bbe 100644 --- a/temporalio/nexus/system/workflow_service/services.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.8. DO NOT EDIT! +# Generated by nexgen v0.2.11. DO NOT EDIT! from __future__ import annotations diff --git a/tests/nexus/test_system_nexus_fallthrough.py b/tests/nexus/test_system_nexus_fallthrough.py new file mode 100644 index 0000000000..f712f9ab2e --- /dev/null +++ b/tests/nexus/test_system_nexus_fallthrough.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import uuid +from collections.abc import Generator +from datetime import timedelta +from typing import Any + +from temporalio import workflow +from temporalio.client import Client +from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT +from temporalio.worker import ( + Interceptor, + StartNexusOperationInput, + WorkflowInboundInterceptor, + WorkflowInterceptorClassInput, + WorkflowOutboundInterceptor, +) +from tests.helpers import new_worker + +PLUGIN_SERVICE = "test.plugin.v1.EchoService" + + +class _DoneHandle(workflow.NexusOperationHandle[Any]): + def __init__(self, value: Any) -> None: + self._value = value + + def cancel(self) -> bool: + return False + + def __await__(self) -> Generator[Any, Any, Any]: + return self._result().__await__() + + async def _result(self) -> Any: + return self._value + + @property + def operation_token(self) -> str | None: + return None + + +class _EchoOutboundInterceptor(WorkflowOutboundInterceptor): + async def start_system_nexus_operation( + self, input: StartNexusOperationInput[Any, Any] + ) -> workflow.NexusOperationHandle[Any]: + if input.service == PLUGIN_SERVICE: + return _DoneHandle(f"{input.operation_name}: {input.input}") + return await super().start_system_nexus_operation(input) + + +class _EchoInboundInterceptor(WorkflowInboundInterceptor): + def init(self, outbound: WorkflowOutboundInterceptor) -> None: + super().init(_EchoOutboundInterceptor(outbound)) + + +class EchoServiceInterceptor(Interceptor): + def workflow_interceptor_class( + self, input: WorkflowInterceptorClassInput + ) -> type[WorkflowInboundInterceptor] | None: + return _EchoInboundInterceptor + + +@workflow.defn +class PluginServiceWorkflow: + @workflow.run + async def run(self) -> str: + client = workflow.create_nexus_client( + service=PLUGIN_SERVICE, endpoint=TEMPORAL_SYSTEM_ENDPOINT + ) + handle: workflow.NexusOperationHandle[Any] = await client.start_operation( + "Echo", "hello" + ) + return await handle + + +async def test_unknown_system_nexus_operation_reaches_generic_hook(client: Client): + async with new_worker( + client, PluginServiceWorkflow, interceptors=[EchoServiceInterceptor()] + ) as worker: + result = await client.execute_workflow( + PluginServiceWorkflow.run, + id=f"wf-{uuid.uuid4()}", + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=10), + ) + assert result == "Echo: hello"