diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index 7ac1bd2de..1e6e6a886 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -11,7 +11,7 @@ base_dir = Path(__file__).parent.parent sys.path.insert(0, str(base_dir)) -wit_input_dir = ( +sdk_core_wit_input_dir = ( base_dir / "temporalio" / "bridge" @@ -22,8 +22,9 @@ / "api_upstream" / "nexus" ) -wit_path = wit_input_dir / "workflow-service.wit" -wit_deps_dir = wit_input_dir / "deps" +# Temporary checked-in WIT used until sdk-core updates its API snapshot. +wit_path = base_dir / "scripts" / "nexus_system_workflow_service.wit" +wit_deps_dir = sdk_core_wit_input_dir / "deps" python_support_path = base_dir / "scripts" / "nex_gen_support.py" output_dir = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" workflow_init_path = base_dir / "temporalio" / "workflow" / "__init__.py" @@ -35,15 +36,28 @@ / "v1" / "request_response.proto" ) +NEX_GEN_VERSION = "0.2.0" def nex_gen_command() -> list[str]: if bin_path := os.environ.get("NEX_GEN_BIN"): return [bin_path] - if shutil.which("nex-gen") is None: - subprocess.check_call(["cargo", "install", "--locked", "nex-gen", "--force"]) - return ["nex-gen"] + if shutil.which("nexgen") is None: + subprocess.check_call( + [ + "cargo", + "install", + "--locked", + "nex-gen", + "--version", + NEX_GEN_VERSION, + "--features", + "advanced", + "--force", + ] + ) + return ["nexgen"] def build_descriptor_set(descriptor_path: Path) -> None: @@ -114,13 +128,10 @@ def generate_nexus_system_api() -> None: subprocess.check_call( [ *command, - "generate", - "--lang", "python", - "--input", str(wit_path), - "--input", str(wit_deps_dir), + "--native-api", "--support-file", str(python_support_path), "--descriptors", diff --git a/scripts/gen_payload_visitor.py b/scripts/gen_payload_visitor.py index 0001659f7..53d61611b 100644 --- a/scripts/gen_payload_visitor.py +++ b/scripts/gen_payload_visitor.py @@ -2,7 +2,7 @@ import sys from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path -from typing import cast +from typing import cast, get_args, get_origin, get_type_hints import google.protobuf.message import nexusrpc @@ -19,14 +19,15 @@ from temporalio.bridge.proto.workflow_completion.workflow_completion_pb2 import ( WorkflowActivationCompletion, ) +from temporalio.converter._payload_converter import _get_transfer_type_converter def discover_system_nexus_roots() -> list[Descriptor]: - module_path = ( - base_dir / "temporalio" / "nexus" / "system" / "workflow_service" / "service.py" - ) + module_path = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" spec = spec_from_file_location( - "temporalio_nexus_system_workflow_service", module_path + "temporalio_nexus_system_workflow_service", + module_path / "__init__.py", + submodule_search_locations=[str(module_path)], ) if spec is None or spec.loader is None: raise RuntimeError(f"Cannot load generated system service from {module_path}") @@ -35,10 +36,14 @@ def discover_system_nexus_roots() -> list[Descriptor]: spec.loader.exec_module(module) roots: list[Descriptor] = [] - for operation in vars(module.WorkflowService).values(): - if not isinstance(operation, nexusrpc.Operation): + for annotation in get_type_hints(module._services.WorkflowService).values(): + if get_origin(annotation) is not nexusrpc.Operation: continue - for proto_type in (operation.input_type, operation.output_type): + for operation_type in get_args(annotation): + converter = _get_transfer_type_converter(operation_type) + proto_type = ( + converter.transfer_type if converter is not None else operation_type + ) if isinstance(proto_type, type) and issubclass( proto_type, google.protobuf.message.Message ): diff --git a/scripts/nex_gen_support.py b/scripts/nex_gen_support.py index 58b51e263..ef6769fa1 100644 --- a/scripts/nex_gen_support.py +++ b/scripts/nex_gen_support.py @@ -10,6 +10,12 @@ import temporalio.api.workflow.v1 import temporalio.common import temporalio.converter +import temporalio.nexus.system + + +class SignalWithStartWorkflowModelRequest(typing.Protocol): + namespace: str + id: str def retry_policy_from_proto( @@ -52,6 +58,12 @@ def workflow_type_to_proto( return common_pb2.WorkflowType(name=workflow_function_name(workflow_type)) +def workflow_type_from_proto( + proto: common_pb2.WorkflowType, +) -> str: + return proto.name + + def task_queue_from_proto( proto: taskqueue_pb2.TaskQueue, ) -> str: @@ -70,12 +82,33 @@ def workflow_namespace() -> str: return info().namespace +def signal_with_start_workflow_serialization_context( + request: SignalWithStartWorkflowModelRequest, +) -> 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: - from temporalio.workflow import payload_converter + return ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + values + ) + ) + - return payload_converter().to_payloads_wrapper(values) +def payloads_from_proto( + proto: common_pb2.Payloads, +) -> list[object]: + return list( + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + proto + ) + ) def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: @@ -87,20 +120,23 @@ def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: def _value_to_payload(value: object | common_pb2.Payload) -> common_pb2.Payload: if isinstance(value, common_pb2.Payload): return _clone_payload(value) - from temporalio.workflow import payload_converter - - payloads = payload_converter().to_payloads_wrapper([value]) + payloads = ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + [value] + ) + ) return _clone_payload(payloads.payloads[0]) def _payload_to_value(payload: common_pb2.Payload) -> object: wrapper = common_pb2.Payloads() wrapper.payloads.add().CopyFrom(payload) - from temporalio.workflow import payload_converter return typing.cast( object, - payload_converter().from_payloads_wrapper(wrapper)[0], + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + wrapper + )[0], ) @@ -177,6 +213,12 @@ def search_attributes_to_proto( return proto +def search_attributes_from_proto( + proto: common_pb2.SearchAttributes, +) -> temporalio.common.TypedSearchAttributes: + return temporalio.converter.decode_typed_search_attributes(proto) + + def priority_from_proto( proto: common_pb2.Priority, ) -> temporalio.common.Priority: @@ -193,3 +235,25 @@ def versioning_override_to_proto( versioning_override: temporalio.common.VersioningOverride, ) -> temporalio.api.workflow.v1.VersioningOverride: return versioning_override._to_proto() # pyright: ignore[reportPrivateUsage] + + +def versioning_override_from_proto( + proto: temporalio.api.workflow.v1.VersioningOverride, +) -> temporalio.common.VersioningOverride: + if proto.HasField("pinned") and proto.pinned.HasField("version"): + version = proto.pinned.version + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion( + deployment_name=version.deployment_name, + build_id=version.build_id, + ) + ) + if proto.pinned_version: + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion.from_canonical_string( + proto.pinned_version + ) + ) + if proto.auto_upgrade: + return temporalio.common.AutoUpgradeVersioningOverride() + raise ValueError("unknown versioning override proto shape") diff --git a/scripts/nexus_system_workflow_service.wit b/scripts/nexus_system_workflow_service.wit new file mode 100644 index 000000000..61822d41c --- /dev/null +++ b/scripts/nexus_system_workflow_service.wit @@ -0,0 +1,128 @@ +package temporal:nexus@1.0.0; + +world system { + export workflow-service; +} + +/// @nexus.endpoint "__temporal_system" +/// @nexus.service-name "temporal.api.workflowservice.v1.WorkflowService" +/// @nexus.namespace dotnet="Temporalio.Workflows" +/// @nexus.operations-class dotnet="Workflow" +/// @nexus.delay-load-temporalio-workflow +/// @nexus.experimental +interface workflow-service { + use nexus:temporal-types/model@1.0.0.{ + duration, + memo, + payloads, + placeholder, + priority, + retry-policy, + search-attributes, + signal-function, + task-queue, + user-metadata, + versioning-override, + workflow-function, + workflow-id-conflict-policy, + workflow-id-reuse-policy, + }; + + /// @nexus.doc "Request fields for signaling a workflow, starting it first if needed." + /// @nexus.experimental + /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest" typescript-import="@temporalio/proto" + record signal-with-start-workflow-request { + /// @nexus.doc + /// python="Workflow type name or callable identifying the workflow to start." + /// typescript="Workflow type name or workflow function identifying the workflow to start." + /// dotnet="Workflow type name or workflow expression identifying the workflow to start." + /// @nexus.proto-field "workflow_type" + workflow: workflow-function, + /// @nexus.doc "Unique identifier for the workflow execution." + /// @nexus.proto-field "workflow_id" + id: string, + /// @nexus.doc "Task queue to run the workflow on." + task-queue: task-queue, + /// @nexus.doc + /// python="Signal name or callable to send with the start request." + /// typescript="Signal name or signal definition to send with the start request." + /// dotnet="Signal name or signal expression to send with the start request." + /// @nexus.proto-field "signal_name" + signal: signal-function, + /// @nexus.doc "Total workflow execution timeout, including retries and continue-as-new." + /// @nexus.proto-field "workflow_execution_timeout" + execution-timeout: option, + /// @nexus.doc "Timeout of a single workflow run." + /// @nexus.proto-field "workflow_run_timeout" + run-timeout: option, + /// @nexus.doc "Timeout of a single workflow task." + /// @nexus.proto-field "workflow_task_timeout" + task-timeout: option, + /// @nexus.omit + identity: placeholder, + /// @nexus.omit + request-id: placeholder, + /// @nexus.doc "Behavior when a closed workflow with the same ID exists. Default is allow-duplicate." + /// @nexus.proto-field "workflow_id_reuse_policy" + /// @nexus.default "allow-duplicate" + id-reuse-policy: workflow-id-reuse-policy, + /// @nexus.doc "Behavior when a workflow is currently running with the same ID. Set to use-existing for idempotent deduplication on workflow ID. Cannot be set if id-reuse-policy is terminate-if-running." + /// @nexus.proto-field "workflow_id_conflict_policy" + id-conflict-policy: option, + /// @nexus.doc "Retry policy for the workflow." + retry-policy: option, + /// @nexus.doc "Cron schedule for recurring workflow executions. See https://docs.temporal.io/cron-job." + cron-schedule: option, + /// @nexus.doc "Memo for the workflow." + memo: option, + /// @nexus.doc "Typed search attributes for the workflow." + search-attributes: option, + /// @nexus.doc "Priority of the workflow execution." + priority: option, + /// @nexus.doc "Override for workflow versioning behavior." + versioning-override: option, + /// @nexus.doc "Amount of time to wait before starting the workflow. This does not work with cron-schedule." + /// @nexus.proto-field "workflow_start_delay" + start-delay: option, + user-metadata: option, + /// @nexus.source python="workflow_namespace" typescript="workflowNamespace" dotnet="TemporalWorkflowContext.WorkflowNamespace" + namespace: string, + /// @nexus.omit + control: placeholder, + /// @nexus.omit + header: placeholder, + /// @nexus.omit + links: placeholder, + /// @nexus.omit + time-skipping-config: placeholder, + } + + /// @nexus.experimental + /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse" typescript-import="@temporalio/proto" + record signal-with-start-workflow-response { + run-id: option, + started: option, + /// @nexus.omit + signal-link: placeholder, + /// @nexus.omit + first-execution-run-id: placeholder, + } + + /// @nexus.doc + /// "Signal a workflow, starting it first if needed." + /// returns="A workflow handle to the started workflow." + /// @nexus.output-transform + /// python-type="temporalio.workflow.ExternalWorkflowHandle[WorkflowResult]" + /// python="temporalio.workflow.get_external_workflow_handle(request.id, run_id=result.run_id)" + /// typescript-type="workflow.ExternalWorkflowHandle" + /// typescript="workflow.getExternalWorkflowHandle(request.id, result.runId ?? undefined)" + /// typescript-import="@temporalio/workflow" + /// dotnet-type="Temporalio.Workflows.ExternalWorkflowHandle" + /// dotnet="Temporalio.Workflows.Workflow.GetExternalWorkflowHandle(request.Id, result.RunId)" + /// @nexus.operation name="SignalWithStartWorkflowExecution" + /// @nexus.serialization-context python="signal_with_start_workflow_serialization_context" + /// @nexus.experimental + signal-with-start-workflow: func( + request: signal-with-start-workflow-request, + ) -> signal-with-start-workflow-response; +} diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index b3d4d2d6f..672c5b14a 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -145,6 +145,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 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..ef6769fa1 100644 --- a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py +++ b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py @@ -10,6 +10,12 @@ import temporalio.api.workflow.v1 import temporalio.common import temporalio.converter +import temporalio.nexus.system + + +class SignalWithStartWorkflowModelRequest(typing.Protocol): + namespace: str + id: str def retry_policy_from_proto( @@ -52,6 +58,12 @@ def workflow_type_to_proto( return common_pb2.WorkflowType(name=workflow_function_name(workflow_type)) +def workflow_type_from_proto( + proto: common_pb2.WorkflowType, +) -> str: + return proto.name + + def task_queue_from_proto( proto: taskqueue_pb2.TaskQueue, ) -> str: @@ -70,12 +82,33 @@ def workflow_namespace() -> str: return info().namespace +def signal_with_start_workflow_serialization_context( + request: SignalWithStartWorkflowModelRequest, +) -> 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: - from temporalio.workflow import payload_converter + return ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + values + ) + ) + - return payload_converter().to_payloads_wrapper(values) +def payloads_from_proto( + proto: common_pb2.Payloads, +) -> list[object]: + return list( + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + proto + ) + ) def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: @@ -87,20 +120,23 @@ def _clone_payload(payload: common_pb2.Payload) -> common_pb2.Payload: def _value_to_payload(value: object | common_pb2.Payload) -> common_pb2.Payload: if isinstance(value, common_pb2.Payload): return _clone_payload(value) - from temporalio.workflow import payload_converter - - payloads = payload_converter().to_payloads_wrapper([value]) + payloads = ( + temporalio.nexus.system._current_user_payload_converter().to_payloads_wrapper( + [value] + ) + ) return _clone_payload(payloads.payloads[0]) def _payload_to_value(payload: common_pb2.Payload) -> object: wrapper = common_pb2.Payloads() wrapper.payloads.add().CopyFrom(payload) - from temporalio.workflow import payload_converter return typing.cast( object, - payload_converter().from_payloads_wrapper(wrapper)[0], + temporalio.nexus.system._current_user_payload_converter().from_payloads_wrapper( + wrapper + )[0], ) @@ -177,6 +213,12 @@ def search_attributes_to_proto( return proto +def search_attributes_from_proto( + proto: common_pb2.SearchAttributes, +) -> temporalio.common.TypedSearchAttributes: + return temporalio.converter.decode_typed_search_attributes(proto) + + def priority_from_proto( proto: common_pb2.Priority, ) -> temporalio.common.Priority: @@ -193,3 +235,25 @@ def versioning_override_to_proto( versioning_override: temporalio.common.VersioningOverride, ) -> temporalio.api.workflow.v1.VersioningOverride: return versioning_override._to_proto() # pyright: ignore[reportPrivateUsage] + + +def versioning_override_from_proto( + proto: temporalio.api.workflow.v1.VersioningOverride, +) -> temporalio.common.VersioningOverride: + if proto.HasField("pinned") and proto.pinned.HasField("version"): + version = proto.pinned.version + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion( + deployment_name=version.deployment_name, + build_id=version.build_id, + ) + ) + if proto.pinned_version: + return temporalio.common.PinnedVersioningOverride( + temporalio.common.WorkerDeploymentVersion.from_canonical_string( + proto.pinned_version + ) + ) + if proto.auto_upgrade: + return temporalio.common.AutoUpgradeVersioningOverride() + raise ValueError("unknown versioning override proto shape") diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index 05e1e3088..d03e7c3b4 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -7,29 +7,196 @@ 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 ( + duration_from_proto, duration_to_proto, + memo_from_proto, memo_to_proto, payload_from_proto, payload_to_proto, + payloads_from_proto, payloads_to_proto, + priority_from_proto, priority_to_proto, + retry_policy_from_proto, retry_policy_to_proto, + search_attributes_from_proto, search_attributes_to_proto, signal_function_to_proto, + task_queue_from_proto, task_queue_to_proto, + versioning_override_from_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_from_proto, 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, + type_hint: type["SignalWithStartWorkflowRequest"], + ) -> "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, + 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, UserMetadata + ) + 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) + ) + 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: """ @@ -46,7 +213,6 @@ class SignalWithStartWorkflowRequest: execution_timeout: datetime.timedelta | None = None run_timeout: datetime.timedelta | None = None task_timeout: datetime.timedelta | None = None - request_id: str | None = None id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ( temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE ) @@ -59,83 +225,101 @@ 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, + type_hint: type["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, + type_hint: type["SignalWithStartWorkflowResponse"], + ) -> "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..ac75f9fb9 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,7 +8,6 @@ import typing_extensions -import temporalio.api.workflowservice.v1.request_response_pb2 import temporalio.common if typing.TYPE_CHECKING: @@ -16,6 +15,7 @@ from ..models import ( SignalWithStartWorkflowRequest, + SignalWithStartWorkflowResponse, UserMetadata, ) @@ -33,15 +33,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) @@ -61,7 +60,6 @@ async def signal_with_start_workflow( execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -77,21 +75,20 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -103,27 +100,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 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -139,24 +134,19 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -168,25 +158,24 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -198,7 +187,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: @@ -217,7 +206,6 @@ async def signal_with_start_workflow( execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -233,24 +221,22 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -266,25 +252,20 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -296,26 +277,26 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -331,23 +312,23 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -381,7 +362,6 @@ async def signal_with_start_workflow( execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -397,26 +377,22 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -432,20 +408,23 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -457,25 +436,28 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -487,27 +469,29 @@ 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 = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -540,7 +524,6 @@ async def signal_with_start_workflow( execution_timeout: datetime.timedelta | None = ..., run_timeout: datetime.timedelta | None = ..., task_timeout: datetime.timedelta | None = ..., - request_id: str | None = ..., id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ..., id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = ..., retry_policy: temporalio.common.RetryPolicy | None = ..., @@ -566,7 +549,6 @@ async def signal_with_start_workflow( execution_timeout: datetime.timedelta | None = None, run_timeout: datetime.timedelta | None = None, task_timeout: datetime.timedelta | None = None, - request_id: str | None = None, id_reuse_policy: temporalio.common.WorkflowIDReusePolicy = ( temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE ), @@ -605,7 +587,6 @@ async def signal_with_start_workflow( continue-as-new. run_timeout: Timeout of a single workflow run. task_timeout: Timeout of a single workflow task. - request_id: Request ID used to deduplicate workflow start requests. id_reuse_policy: Behavior when a closed workflow with the same ID exists. Default is allow-duplicate. id_conflict_policy: Behavior when a workflow is currently running with the same @@ -629,6 +610,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 +622,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 @@ -659,7 +640,6 @@ async def signal_with_start_workflow( execution_timeout=execution_timeout, run_timeout=run_timeout, task_timeout=task_timeout, - request_id=request_id, id_reuse_policy=id_reuse_policy, id_conflict_policy=id_conflict_policy, retry_policy=retry_policy, 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 d0b10ccae..1e13635c7 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -1032,10 +1032,23 @@ def _apply_resolve_nexus_operation( # Handle the four oneof variants of NexusOperationResult result = job.result if result.HasField("completed"): + payload_converter = handle._payload_converter + if temporalio.nexus.system.is_system_endpoint(handle._input.endpoint): + serialization_context = ( + temporalio.nexus.system._get_serialization_context( + handle._input.service, + handle._input.operation_name, + handle._input.input, + ) + ) + if serialization_context is not None: + payload_converter = temporalio.nexus.system._get_payload_converter( + self._payload_converter_with_context(serialization_context) + ) [output] = self._convert_payloads( [result.completed], [handle._input.output_type] if handle._input.output_type else None, - handle._payload_converter, + payload_converter, ) handle._resolve_success(output) elif result.HasField("failed"): @@ -2133,13 +2146,12 @@ async def operation_handle_fn() -> OutputT: ), ) - 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"), @@ -2382,9 +2394,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 ec86689d7..52ac75516 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -14,6 +14,8 @@ import temporalio.api.workflowservice.v1.request_response_pb2 as workflowservice_pb2 import temporalio.converter import temporalio.nexus.system as nexus_system +import temporalio.nexus.system.workflow_service as workflow_service +import temporalio.nexus.system.workflow_service.models as workflow_service_models from temporalio import workflow from temporalio.bridge._visitor import PayloadVisitor from temporalio.bridge._visitor_functions import VisitorFunctions @@ -21,7 +23,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, @@ -56,6 +64,32 @@ async def run(self, task_queue: str) -> str: return started_handle.id +def test_signal_with_start_serialization_context() -> None: + request = workflow_service_models.SignalWithStartWorkflowRequest( + workflow="test-workflow", + id="target-workflow-id", + task_queue="target-task-queue", + signal="test-signal", + namespace="target-namespace", + ) + operation_info = workflow_service.__nexus_operation_registry__[ + ( + "temporal.api.workflowservice.v1.WorkflowService", + "SignalWithStartWorkflowExecution", + ) + ] + + assert operation_info.serialization_context is not None + context = nexus_system._get_serialization_context( + "temporal.api.workflowservice.v1.WorkflowService", + "SignalWithStartWorkflowExecution", + request, + ) + assert isinstance(context, WorkflowSerializationContext) + assert context.namespace == "target-namespace" + assert context.workflow_id == "target-workflow-id" + + class RejectOuterSystemNexusCodec(PayloadCodec): def __init__(self) -> None: self.encode_count = 0 @@ -99,6 +133,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 @@ -136,13 +198,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(VisitorFunctions): @@ -426,3 +485,57 @@ 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, + monkeypatch: pytest.MonkeyPatch, +) -> None: + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + + captured_contexts: list[SerializationContext | None] = [] + system_payload_converter_wrap_count = 0 + original_get_payload_converter = nexus_system._get_payload_converter + + def capture_get_payload_converter( + user_payload_converter: temporalio.converter.PayloadConverter, + ) -> temporalio.converter.PayloadConverter: + nonlocal system_payload_converter_wrap_count + system_payload_converter_wrap_count += 1 + return original_get_payload_converter(user_payload_converter) + + monkeypatch.setattr( + nexus_system, "_get_payload_converter", capture_get_payload_converter + ) + 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 system_payload_converter_wrap_count >= 2 + assert len(captured_contexts) >= 2 + assert all( + isinstance(context, WorkflowSerializationContext) + and context.workflow_id == target_workflow_id + for context in captured_contexts + )