diff --git a/scripts/nex_gen_support.py b/scripts/nex_gen_support.py index 58b51e263..6b98ee840 100644 --- a/scripts/nex_gen_support.py +++ b/scripts/nex_gen_support.py @@ -12,6 +12,11 @@ import temporalio.converter +class SignalWithStartWorkflowRequest(typing.Protocol): + namespace: str + id: str + + def retry_policy_from_proto( proto: common_pb2.RetryPolicy, ) -> temporalio.common.RetryPolicy: @@ -70,6 +75,15 @@ def workflow_namespace() -> str: return info().namespace +def signal_with_start_workflow_serialization_context( + request: SignalWithStartWorkflowRequest, +) -> temporalio.converter.WorkflowSerializationContext: + return temporalio.converter.WorkflowSerializationContext( + namespace=request.namespace, + workflow_id=request.id, + ) + + def payloads_to_proto( values: collections.abc.Sequence[typing.Any], ) -> common_pb2.Payloads: diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 14a43cb72..21d39e79f 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -121,6 +121,20 @@ def _get_payload_converter( # pyright: ignore[reportUnusedFunction] return _SystemNexusPayloadConverter(user_payload_converter) +def _get_serialization_context( # pyright: ignore[reportUnusedFunction] + service: str, + operation: str, + request: Any, +) -> temporalio.converter.SerializationContext | None: + """Return the target serialization context for a system Nexus operation.""" + from .workflow_service import __nexus_operation_registry__ + + operation_info = __nexus_operation_registry__.get((service, operation)) + if operation_info is None or operation_info.serialization_context is None: + return None + return operation_info.serialization_context(request) + + __all__ = [ "TEMPORAL_SYSTEM_ENDPOINT", "is_system_endpoint", diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index 7c24fa125..092383b35 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -2,7 +2,15 @@ from __future__ import annotations -from . import service as _service +import collections.abc +import typing + +import nexusrpc + +import temporalio.converter + +from . import services as _services +from ._support import signal_with_start_workflow_serialization_context from .operations.signal_with_start_workflow import signal_with_start_workflow __all__ = [ @@ -10,9 +18,34 @@ ] +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +_SerializationContextFactory = collections.abc.Callable[ + [_InputT], temporalio.converter.SerializationContext +] + + +class _NexusOperationInfo(typing.Generic[_InputT, _OutputT]): + def __init__( + self, + *, + operation: nexusrpc.Operation[_InputT, _OutputT], + serialization_context: _SerializationContextFactory[_InputT] | None = None, + ) -> None: + self.operation: nexusrpc.Operation[_InputT, _OutputT] = operation + self.serialization_context: _SerializationContextFactory[_InputT] | None = ( + serialization_context + ) + + __nexus_operation_registry__ = { ( "temporal.api.workflowservice.v1.WorkflowService", "SignalWithStartWorkflowExecution", - ): _service.WorkflowService.signal_with_start_workflow, + ): _NexusOperationInfo( + operation=_services.WorkflowService.signal_with_start_workflow, + serialization_context=signal_with_start_workflow_serialization_context, + ), } diff --git a/temporalio/nexus/system/workflow_service/_resources/__init__.py b/temporalio/nexus/system/workflow_service/_resources/__init__.py deleted file mode 100644 index 373efbd33..000000000 --- a/temporalio/nexus/system/workflow_service/_resources/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -# Generated by nex-gen. DO NOT EDIT! - -from __future__ import annotations - -__all__ = [] diff --git a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py index 58b51e263..6b98ee840 100644 --- a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py +++ b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py @@ -12,6 +12,11 @@ import temporalio.converter +class SignalWithStartWorkflowRequest(typing.Protocol): + namespace: str + id: str + + def retry_policy_from_proto( proto: common_pb2.RetryPolicy, ) -> temporalio.common.RetryPolicy: @@ -70,6 +75,15 @@ def workflow_namespace() -> str: return info().namespace +def signal_with_start_workflow_serialization_context( + request: SignalWithStartWorkflowRequest, +) -> temporalio.converter.WorkflowSerializationContext: + return temporalio.converter.WorkflowSerializationContext( + namespace=request.namespace, + workflow_id=request.id, + ) + + def payloads_to_proto( values: collections.abc.Sequence[typing.Any], ) -> common_pb2.Payloads: diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index 05e1e3088..c690abd62 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -7,29 +7,195 @@ import datetime import typing +import typing_extensions + import temporalio.api.sdk.v1.user_metadata_pb2 import temporalio.api.workflowservice.v1.request_response_pb2 import temporalio.common +import temporalio.converter from ._support import ( + SignalWithStartWorkflowRequest, + duration_from_proto, duration_to_proto, + memo_from_proto, memo_to_proto, payload_from_proto, payload_to_proto, payloads_to_proto, + priority_from_proto, priority_to_proto, + retry_policy_from_proto, retry_policy_to_proto, search_attributes_to_proto, signal_function_to_proto, + task_queue_from_proto, task_queue_to_proto, versioning_override_to_proto, + workflow_id_conflict_policy_from_proto, workflow_id_conflict_policy_to_proto, + workflow_id_reuse_policy_from_proto, workflow_id_reuse_policy_to_proto, workflow_namespace, workflow_type_to_proto, ) +class _SignalWithStartWorkflowRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + "SignalWithStartWorkflowRequest", + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, + ] +): + transfer_type: ( + type[ + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest + ] + | None + ) = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, + ) -> "SignalWithStartWorkflowRequest": + proto = value + if not proto.HasField("workflow_type"): + raise ValueError( + "missing required field SignalWithStartWorkflowRequest.workflow" + ) + workflow = workflow_type_from_proto(proto.workflow_type) + if not proto.workflow_id: + raise ValueError("missing required field SignalWithStartWorkflowRequest.id") + id = proto.workflow_id + if not proto.HasField("task_queue"): + raise ValueError( + "missing required field SignalWithStartWorkflowRequest.task_queue" + ) + task_queue = task_queue_from_proto(proto.task_queue) + if not proto.signal_name: + raise ValueError( + "missing required field SignalWithStartWorkflowRequest.signal" + ) + signal = proto.signal_name + return SignalWithStartWorkflowRequest( + workflow=workflow, + args=payloads_from_proto(proto.input) if proto.HasField("input") else None, + id=id, + task_queue=task_queue, + signal=signal, + signal_args=payloads_from_proto(proto.signal_input) + if proto.HasField("signal_input") + else None, + execution_timeout=duration_from_proto(proto.workflow_execution_timeout) + if proto.HasField("workflow_execution_timeout") + else None, + run_timeout=duration_from_proto(proto.workflow_run_timeout) + if proto.HasField("workflow_run_timeout") + else None, + task_timeout=duration_from_proto(proto.workflow_task_timeout) + if proto.HasField("workflow_task_timeout") + else None, + request_id=proto.request_id if bool(proto.request_id) else None, + id_reuse_policy=workflow_id_reuse_policy_from_proto( + proto.workflow_id_reuse_policy + ), + id_conflict_policy=workflow_id_conflict_policy_from_proto( + proto.workflow_id_conflict_policy + ) + if proto.workflow_id_conflict_policy != 0 + else None, + retry_policy=retry_policy_from_proto(proto.retry_policy) + if proto.HasField("retry_policy") + else None, + cron_schedule=proto.cron_schedule if bool(proto.cron_schedule) else None, + memo=memo_from_proto(proto.memo) if proto.HasField("memo") else None, + search_attributes=search_attributes_from_proto(proto.search_attributes) + if proto.HasField("search_attributes") + else None, + priority=priority_from_proto(proto.priority) + if proto.HasField("priority") + else None, + versioning_override=versioning_override_from_proto( + proto.versioning_override + ) + if proto.HasField("versioning_override") + else None, + start_delay=duration_from_proto(proto.workflow_start_delay) + if proto.HasField("workflow_start_delay") + else None, + user_metadata=_UserMetadataTransferTypeConverter().from_transfer_type( + proto.user_metadata + ) + if proto.HasField("user_metadata") + else None, + namespace=proto.namespace, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: "SignalWithStartWorkflowRequest", + ) -> temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest: + message = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest() + message.workflow_type.CopyFrom(workflow_type_to_proto(value.workflow)) + if value.args is not None: + message.input.CopyFrom(payloads_to_proto(value.args)) + message.workflow_id = value.id + message.task_queue.CopyFrom(task_queue_to_proto(value.task_queue)) + message.signal_name = signal_function_to_proto(value.signal) + if value.signal_args is not None: + message.signal_input.CopyFrom(payloads_to_proto(value.signal_args)) + if value.execution_timeout is not None: + message.workflow_execution_timeout.CopyFrom( + duration_to_proto(value.execution_timeout) + ) + if value.run_timeout is not None: + message.workflow_run_timeout.CopyFrom(duration_to_proto(value.run_timeout)) + if value.task_timeout is not None: + message.workflow_task_timeout.CopyFrom( + duration_to_proto(value.task_timeout) + ) + if value.request_id is not None: + message.request_id = value.request_id + message.workflow_id_reuse_policy = workflow_id_reuse_policy_to_proto( + value.id_reuse_policy + ) + if value.id_conflict_policy is not None: + message.workflow_id_conflict_policy = workflow_id_conflict_policy_to_proto( + value.id_conflict_policy + ) + if value.retry_policy is not None: + message.retry_policy.CopyFrom(retry_policy_to_proto(value.retry_policy)) + if value.cron_schedule is not None: + message.cron_schedule = value.cron_schedule + if value.memo is not None: + message.memo.CopyFrom(memo_to_proto(value.memo)) + if value.search_attributes is not None: + message.search_attributes.CopyFrom( + search_attributes_to_proto(value.search_attributes) + ) + if value.priority is not None: + message.priority.CopyFrom(priority_to_proto(value.priority)) + if value.versioning_override is not None: + message.versioning_override.CopyFrom( + versioning_override_to_proto(value.versioning_override) + ) + if value.start_delay is not None: + message.workflow_start_delay.CopyFrom(duration_to_proto(value.start_delay)) + if value.user_metadata is not None: + message.user_metadata.CopyFrom( + _UserMetadataTransferTypeConverter().to_transfer_type( + value.user_metadata + ) + ) + message.namespace = value.namespace + return message + + +@temporalio.converter.transfer_type_convertible( + _SignalWithStartWorkflowRequestTransferTypeConverter +) @dataclasses.dataclass(slots=True, kw_only=True) class SignalWithStartWorkflowRequest: """ @@ -59,83 +225,99 @@ class SignalWithStartWorkflowRequest: versioning_override: temporalio.common.VersioningOverride | None = None start_delay: datetime.timedelta | None = None user_metadata: UserMetadata | None = None + namespace: str = dataclasses.field(default_factory=workflow_namespace) + - def to_proto( +class _UserMetadataTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + "UserMetadata", temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata + ] +): + transfer_type: type[temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata] | None = ( + temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata + ) + + @typing_extensions.override + def from_transfer_type( self, - ) -> temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest: - message = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest() - message.workflow_type.CopyFrom(workflow_type_to_proto(self.workflow)) - if self.args is not None: - message.input.CopyFrom(payloads_to_proto(self.args)) - message.workflow_id = self.id - message.task_queue.CopyFrom(task_queue_to_proto(self.task_queue)) - message.signal_name = signal_function_to_proto(self.signal) - if self.signal_args is not None: - message.signal_input.CopyFrom(payloads_to_proto(self.signal_args)) - if self.execution_timeout is not None: - message.workflow_execution_timeout.CopyFrom( - duration_to_proto(self.execution_timeout) - ) - if self.run_timeout is not None: - message.workflow_run_timeout.CopyFrom(duration_to_proto(self.run_timeout)) - if self.task_timeout is not None: - message.workflow_task_timeout.CopyFrom(duration_to_proto(self.task_timeout)) - if self.request_id is not None: - message.request_id = self.request_id - message.workflow_id_reuse_policy = workflow_id_reuse_policy_to_proto( - self.id_reuse_policy + value: temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata, + ) -> "UserMetadata": + proto = value + return UserMetadata( + static_summary=payload_from_proto(proto.summary) + if proto.HasField("summary") + else None, + static_details=payload_from_proto(proto.details) + if proto.HasField("details") + else None, ) - if self.id_conflict_policy is not None: - message.workflow_id_conflict_policy = workflow_id_conflict_policy_to_proto( - self.id_conflict_policy - ) - if self.retry_policy is not None: - message.retry_policy.CopyFrom(retry_policy_to_proto(self.retry_policy)) - if self.cron_schedule is not None: - message.cron_schedule = self.cron_schedule - if self.memo is not None: - message.memo.CopyFrom(memo_to_proto(self.memo)) - if self.search_attributes is not None: - message.search_attributes.CopyFrom( - search_attributes_to_proto(self.search_attributes) - ) - if self.priority is not None: - message.priority.CopyFrom(priority_to_proto(self.priority)) - if self.versioning_override is not None: - message.versioning_override.CopyFrom( - versioning_override_to_proto(self.versioning_override) - ) - if self.start_delay is not None: - message.workflow_start_delay.CopyFrom(duration_to_proto(self.start_delay)) - if self.user_metadata is not None: - message.user_metadata.CopyFrom(self.user_metadata.to_proto()) - message.namespace = workflow_namespace() + + @typing_extensions.override + def to_transfer_type( + self, + value: "UserMetadata", + ) -> temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata: + message = temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata() + if value.static_summary is not None: + message.summary.CopyFrom(payload_to_proto(value.static_summary)) + if value.static_details is not None: + message.details.CopyFrom(payload_to_proto(value.static_details)) return message +@temporalio.converter.transfer_type_convertible(_UserMetadataTransferTypeConverter) @dataclasses.dataclass(slots=True) class UserMetadata: static_summary: typing.Any | None = None static_details: typing.Any | None = None - @classmethod - def from_proto( - cls, - proto: temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata, - ) -> UserMetadata: - return cls( - static_summary=payload_from_proto(proto.summary) - if proto.HasField("summary") - else None, - static_details=payload_from_proto(proto.details) - if proto.HasField("details") - else None, + +class _SignalWithStartWorkflowResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + "SignalWithStartWorkflowResponse", + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + ] +): + transfer_type: ( + type[ + temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse + ] + | None + ) = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + ) -> "SignalWithStartWorkflowResponse": + proto = value + return SignalWithStartWorkflowResponse( + run_id=proto.run_id if bool(proto.run_id) else None, + started=proto.started if bool(proto.started) else None, ) - def to_proto(self) -> temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata: - message = temporalio.api.sdk.v1.user_metadata_pb2.UserMetadata() - if self.static_summary is not None: - message.summary.CopyFrom(payload_to_proto(self.static_summary)) - if self.static_details is not None: - message.details.CopyFrom(payload_to_proto(self.static_details)) + @typing_extensions.override + def to_transfer_type( + self, + value: "SignalWithStartWorkflowResponse", + ) -> temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse: + message = temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse() + if value.run_id is not None: + message.run_id = value.run_id + if value.started is not None: + message.started = value.started return message + + +@temporalio.converter.transfer_type_convertible( + _SignalWithStartWorkflowResponseTransferTypeConverter +) +@dataclasses.dataclass(slots=True) +class SignalWithStartWorkflowResponse: + """ + .. warning:: + This API is experimental and subject to change. + """ + + run_id: str | None = None + started: bool | None = None 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 0865e5a88..5f8e7b64e 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 @@ -8,14 +8,15 @@ import typing_extensions -import temporalio.api.workflowservice.v1.request_response_pb2 import temporalio.common if typing.TYPE_CHECKING: from temporalio.workflow import ExternalWorkflowHandle +from .._support import SignalWithStartWorkflowRequest from ..models import ( SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, UserMetadata, ) @@ -33,15 +34,14 @@ async def _signal_with_start_workflow( get_external_workflow_handle, ) - request_proto = request.to_proto() nexus_client = create_nexus_client( service="temporal.api.workflowservice.v1.WorkflowService", endpoint="__temporal_system", ) handle = await nexus_client.start_operation( operation="SignalWithStartWorkflowExecution", - input=request_proto, - output_type=temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + input=request, + output_type=SignalWithStartWorkflowResponse, ) result = await handle return get_external_workflow_handle(request.id, run_id=result.run_id) @@ -77,17 +77,17 @@ async def signal_with_start_workflow( # Overload case: -# - workflow name with optional list-form workflow arguments -# - signal name with optional list-form signal arguments +# - workflow name with positional workflow arguments +# - signal method callable with no signal arguments @typing.overload async def signal_with_start_workflow( workflow: str, - *, - args: list[typing.Any] | None = ..., + *args: object, id: str, task_queue: str, - signal: str, - signal_args: list[typing.Any] | None = ..., + signal: collections.abc.Callable[ + [SelfType], None | collections.abc.Awaitable[None] + ], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -103,23 +103,22 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[object]: ... +) -> ExternalWorkflowHandle[SelfType]: ... # Overload case: -# - workflow method callable with typed positional workflow arguments -# - signal name with optional list-form signal arguments +# - workflow name with positional workflow arguments +# - signal method callable with a typed single signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], - *args: typing_extensions.Unpack[WorkflowArgs], + workflow: str, + *args: object, id: str, task_queue: str, - signal: str, - signal_args: list[typing.Any] | None = ..., + signal: collections.abc.Callable[ + [SelfType, SignalArg], None | collections.abc.Awaitable[None] + ], + signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -139,20 +138,16 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with list-form workflow arguments -# - signal name with optional list-form signal arguments +# - workflow name with positional workflow arguments +# - signal callable with list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], - *, - args: list[typing.Any], + workflow: str, + *args: object, id: str, task_queue: str, - signal: str, - signal_args: list[typing.Any] | None = ..., + signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], + signal_args: list[typing.Any], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -168,21 +163,21 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[SelfType]: ... +) -> ExternalWorkflowHandle[object]: ... # Overload case: -# - workflow name with positional workflow arguments -# - signal method callable with no signal arguments +# - workflow name with optional list-form workflow arguments +# - signal name with optional list-form signal arguments @typing.overload async def signal_with_start_workflow( workflow: str, - *args: object, + *, + args: list[typing.Any] | None = ..., id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType], None | collections.abc.Awaitable[None] - ], + signal: str, + signal_args: list[typing.Any] | None = ..., execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -198,7 +193,7 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[SelfType]: ... +) -> ExternalWorkflowHandle[object]: ... # Overload case: @@ -233,20 +228,19 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with typed positional workflow arguments -# - signal method callable with no signal arguments +# - workflow name with optional list-form workflow arguments +# - signal method callable with a typed single signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], - *args: typing_extensions.Unpack[WorkflowArgs], + workflow: str, + *, + args: list[typing.Any] | None = ..., id: str, task_queue: str, signal: collections.abc.Callable[ - [SelfType], None | collections.abc.Awaitable[None] + [SelfType, SignalArg], None | collections.abc.Awaitable[None] ], + signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -266,21 +260,17 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with list-form workflow arguments -# - signal method callable with no signal arguments +# - workflow name with optional list-form workflow arguments +# - signal callable with list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: collections.abc.Callable[ - [SelfType, typing_extensions.Unpack[WorkflowArgs]], - collections.abc.Awaitable[WorkflowResult], - ], + workflow: str, *, - args: list[typing.Any], + args: list[typing.Any] | None = ..., id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType], None | collections.abc.Awaitable[None] - ], + signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], + signal_args: list[typing.Any], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -296,22 +286,23 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[SelfType]: ... +) -> ExternalWorkflowHandle[object]: ... # Overload case: -# - workflow name with positional workflow arguments -# - signal method callable with a typed single signal arguments +# - workflow method callable with typed positional workflow arguments +# - signal name with optional list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, - *args: object, + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], + *args: typing_extensions.Unpack[WorkflowArgs], id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType, SignalArg], None | collections.abc.Awaitable[None] - ], - signal_args: SignalArg, + signal: str, + signal_args: list[typing.Any] | None = ..., execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -331,19 +322,20 @@ async def signal_with_start_workflow( # Overload case: -# - workflow name with optional list-form workflow arguments -# - signal method callable with a typed single signal arguments +# - workflow method callable with typed positional workflow arguments +# - signal method callable with no signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, - *, - args: list[typing.Any] | None = ..., + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], + *args: typing_extensions.Unpack[WorkflowArgs], id: str, task_queue: str, signal: collections.abc.Callable[ - [SelfType, SignalArg], None | collections.abc.Awaitable[None] + [SelfType], None | collections.abc.Awaitable[None] ], - signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -397,22 +389,19 @@ async def signal_with_start_workflow( # Overload case: -# - workflow method callable with list-form workflow arguments -# - signal method callable with a typed single signal arguments +# - workflow method callable with typed positional workflow arguments +# - signal callable with list-form signal arguments @typing.overload async def signal_with_start_workflow( workflow: collections.abc.Callable[ [SelfType, typing_extensions.Unpack[WorkflowArgs]], collections.abc.Awaitable[WorkflowResult], ], - *, - args: list[typing.Any], + *args: typing_extensions.Unpack[WorkflowArgs], id: str, task_queue: str, - signal: collections.abc.Callable[ - [SelfType, SignalArg], None | collections.abc.Awaitable[None] - ], - signal_args: SignalArg, + signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], + signal_args: list[typing.Any], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -432,16 +421,20 @@ async def signal_with_start_workflow( # Overload case: -# - workflow name with positional workflow arguments -# - signal callable with list-form signal arguments +# - workflow method callable with list-form workflow arguments +# - signal name with optional list-form signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, - *args: object, + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], + *, + args: list[typing.Any], id: str, task_queue: str, - signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], - signal_args: list[typing.Any], + signal: str, + signal_args: list[typing.Any] | None = ..., execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -457,21 +450,25 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[object]: ... +) -> ExternalWorkflowHandle[SelfType]: ... # Overload case: -# - workflow name with optional list-form workflow arguments -# - signal callable with list-form signal arguments +# - workflow method callable with list-form workflow arguments +# - signal method callable with no signal arguments @typing.overload async def signal_with_start_workflow( - workflow: str, + workflow: collections.abc.Callable[ + [SelfType, typing_extensions.Unpack[WorkflowArgs]], + collections.abc.Awaitable[WorkflowResult], + ], *, - args: list[typing.Any] | None = ..., + args: list[typing.Any], id: str, task_queue: str, - signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], - signal_args: list[typing.Any], + signal: collections.abc.Callable[ + [SelfType], None | collections.abc.Awaitable[None] + ], execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -487,23 +484,26 @@ async def signal_with_start_workflow( start_delay: datetime.timedelta | None = ..., static_summary: str | None = ..., static_details: str | None = ..., -) -> ExternalWorkflowHandle[object]: ... +) -> ExternalWorkflowHandle[SelfType]: ... # Overload case: -# - workflow method callable with typed positional workflow arguments -# - signal callable with list-form signal arguments +# - workflow method callable with list-form workflow arguments +# - signal method callable with a typed single signal arguments @typing.overload async def signal_with_start_workflow( workflow: collections.abc.Callable[ [SelfType, typing_extensions.Unpack[WorkflowArgs]], collections.abc.Awaitable[WorkflowResult], ], - *args: typing_extensions.Unpack[WorkflowArgs], + *, + args: list[typing.Any], id: str, task_queue: str, - signal: collections.abc.Callable[..., None | collections.abc.Awaitable[None]], - signal_args: list[typing.Any], + signal: collections.abc.Callable[ + [SelfType, SignalArg], None | collections.abc.Awaitable[None] + ], + signal_args: SignalArg, execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., @@ -629,6 +629,11 @@ async def signal_with_start_workflow( Returns: A workflow handle to the started workflow. """ + if positional_args and args is not None: + raise TypeError("cannot specify both positional arguments and args") + normalized_args: list[typing.Any] | None = ( + list(positional_args) if positional_args else args + ) normalized_signal_args: list[typing.Any] | None if signal_args is None: normalized_signal_args = None @@ -636,11 +641,6 @@ async def signal_with_start_workflow( normalized_signal_args = typing.cast(list[typing.Any], signal_args) else: normalized_signal_args = [signal_args] - if positional_args and args is not None: - raise TypeError("cannot specify both positional arguments and args") - normalized_args: list[typing.Any] | None = ( - list(positional_args) if positional_args else args - ) user_metadata = ( None if static_summary is None and static_details is None diff --git a/temporalio/nexus/system/workflow_service/service.py b/temporalio/nexus/system/workflow_service/services.py similarity index 63% rename from temporalio/nexus/system/workflow_service/service.py rename to temporalio/nexus/system/workflow_service/services.py index 7ce5849ca..e2e901157 100644 --- a/temporalio/nexus/system/workflow_service/service.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -4,7 +4,10 @@ from nexusrpc import Operation, service -import temporalio.api.workflowservice.v1.request_response_pb2 +from .models import ( + SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, +) @service(name="temporal.api.workflowservice.v1.WorkflowService") @@ -16,6 +19,6 @@ class WorkflowService: # .. warning:: This API is experimental and subject to change. signal_with_start_workflow: Operation[ - temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionRequest, - temporalio.api.workflowservice.v1.request_response_pb2.SignalWithStartWorkflowExecutionResponse, + SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, ] = Operation(name="SignalWithStartWorkflowExecution") diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 76b3304ef..da563f32f 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2129,13 +2129,12 @@ async def operation_handle_fn() -> OutputT: ): t.uncancel() # type: ignore[union-attr] - payload_converter = ( - temporalio.nexus.system._get_payload_converter( + if temporalio.nexus.system.is_system_endpoint(input.endpoint): + payload_converter = temporalio.nexus.system._get_payload_converter( self._workflow_context_payload_converter ) - if temporalio.nexus.system.is_system_endpoint(input.endpoint) - else self._context_free_payload_converter - ) + else: + payload_converter = self._context_free_payload_converter handle = _NexusOperationHandle( self, self._next_seq("nexus_operation"), @@ -2333,9 +2332,17 @@ def get_serialization_context( == temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION and command_info.command_seq in self._pending_nexus_operations ): - # Use empty context for nexus operations: users will never want to encrypt using a - # key derived from caller workflow context because the caller workflow context is - # not available on the handler side for decryption. + nexus_operation = self._pending_nexus_operations[command_info.command_seq] + if temporalio.nexus.system.is_system_endpoint( + nexus_operation._input.endpoint + ): + return temporalio.nexus.system._get_serialization_context( + nexus_operation._input.service, + nexus_operation._input.operation_name, + nexus_operation._input.input, + ) + # Other Nexus operations have no context because the caller workflow context is + # unavailable on the handler side for decryption. return None else: diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index eb8ee603c..f279ddab6 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -20,7 +20,13 @@ WorkflowActivationCompletion, ) from temporalio.client import Client -from temporalio.converter import ExternalStorage, PayloadCodec +from temporalio.converter import ( + ExternalStorage, + PayloadCodec, + SerializationContext, + WithSerializationContext, + WorkflowSerializationContext, +) from temporalio.testing import WorkflowEnvironment from temporalio.worker import ( Interceptor, @@ -97,6 +103,34 @@ async def decode( return decoded +class CaptureSystemNexusPayloadContextCodec(PayloadCodec, WithSerializationContext): + def __init__( + self, + captured_contexts: list[SerializationContext | None], + context: SerializationContext | None = None, + ) -> None: + self._captured_contexts = captured_contexts + self._context = context + + def with_context( + self, context: SerializationContext + ) -> CaptureSystemNexusPayloadContextCodec: + return CaptureSystemNexusPayloadContextCodec(self._captured_contexts, context) + + async def encode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + for payload in payloads: + if payload.data in {b'"workflow-input"', b'"signal-input"'}: + self._captured_contexts.append(self._context) + return list(payloads) + + async def decode( + self, payloads: Sequence[temporalio.api.common.v1.Payload] + ) -> list[temporalio.api.common.v1.Payload]: + return list(payloads) + + class TracingWorkflowInterceptor(Interceptor): def workflow_interceptor_class( self, input: WorkflowInterceptorClassInput @@ -134,13 +168,10 @@ def _assert_start_nexus_operation_interceptor_trace() -> None: trace_name, trace_value = interceptor_traces.pop() assert trace_name == "workflow.start_nexus_operation" trace_input = cast(StartNexusOperationInput[Any, Any], trace_value) - request = cast( - workflowservice_pb2.SignalWithStartWorkflowExecutionRequest, - trace_input.input, - ) - assert request.workflow_id == "system-nexus-workflow-id" - assert request.signal_name == "test-signal" - assert request.workflow_type.name == "test-workflow" + request = trace_input.input + assert request.id == "system-nexus-workflow-id" + assert request.signal == "test-signal" + assert request.workflow == "test-workflow" class _MarkingPayloadVisitor: @@ -408,3 +439,42 @@ async def test_external_workflow_handle_signal_with_start_workflow_uses_system_n }, ) _assert_start_nexus_operation_interceptor_trace() + + +async def test_signal_with_start_uses_target_workflow_serialization_context( + env: WorkflowEnvironment, +) -> None: + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + captured_contexts: list[SerializationContext | None] = [] + caller_config = env.client.config() + caller_config["data_converter"] = dataclasses.replace( + temporalio.converter.default(), + payload_codec=CaptureSystemNexusPayloadContextCodec(captured_contexts), + ) + caller_client = Client(**caller_config) + caller_task_queue = str(uuid.uuid4()) + target_workflow_id = "system-nexus-workflow-id" + + async with Worker( + caller_client, + task_queue=caller_task_queue, + workflows=[ExternalHandleSignalWithStartWorkflowCaller], + workflow_runner=UnsandboxedWorkflowRunner(), + ): + result = await caller_client.execute_workflow( + ExternalHandleSignalWithStartWorkflowCaller.run, + args=[str(uuid.uuid4())], + id=str(uuid.uuid4()), + task_queue=caller_task_queue, + execution_timeout=timedelta(seconds=5), + ) + + assert result == target_workflow_id + assert len(captured_contexts) >= 2 + assert all( + isinstance(context, WorkflowSerializationContext) + and context.workflow_id == target_workflow_id + for context in captured_contexts + )