diff --git a/pyproject.toml b/pyproject.toml index 400319165..845e5f980 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -205,7 +205,7 @@ ignore_errors = true [tool.pydocstyle] convention = "google" # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088 -match_dir = "^(?!(docs|scripts|tests|api|proto|system|\\.)).*" +match_dir = "^(?!(docs|scripts|tests|api|proto|system|_support|notifications|\\.)).*" add_ignore = [ # We like to wrap at a certain number of chars, even long summary sentences. # https://github.com/PyCQA/pydocstyle/issues/184 diff --git a/scripts/gen_nexus_system_api.py b/scripts/gen_nexus_system_api.py index e57ca03b3..d11240073 100644 --- a/scripts/gen_nexus_system_api.py +++ b/scripts/gen_nexus_system_api.py @@ -1,11 +1,11 @@ +import ast import os +import re import shutil import subprocess import sys import tempfile -from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path -from typing import cast import gen_protos @@ -22,27 +22,37 @@ / "api_upstream" / "nexus" ) -wit_path = wit_input_dir / "workflow-service.wit" wit_deps_dir = wit_input_dir / "deps" python_support_path = base_dir / "scripts" / "nex_gen_support.py" -output_dir = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" +workflow_output_dir = base_dir / "temporalio" / "nexus" / "system" / "workflow_service" +notification_output_dir = base_dir / "temporalio" / "nexus" / "notifications" +support_output_dir = base_dir / "temporalio" / "nexus" / "_support" workflow_init_path = base_dir / "temporalio" / "workflow" / "__init__.py" -workflowservice_request_response_proto = ( +NEX_GEN_VERSION = "0.2.7" +proto_files = [ gen_protos.api_proto_dir / "temporal" / "api" / "workflowservice" / "v1" - / "request_response.proto" -) -NEX_GEN_VERSION = "0.2.4" + / "request_response.proto", + gen_protos.api_proto_dir + / "temporal" + / "api" + / "notificationservice" + / "v1" + / "request_response.proto", +] def nex_gen_command() -> list[str]: if bin_path := os.environ.get("NEX_GEN_BIN"): return [bin_path] - - if shutil.which("nexgen") is None: + if ( + shutil.which("nexgen") is None + or subprocess.check_output(["nexgen", "--version"], text=True).strip() + != f"nexgen {NEX_GEN_VERSION}" + ): subprocess.check_call( [ "cargo", @@ -68,24 +78,99 @@ def build_descriptor_set(descriptor_path: Path) -> None: f"--proto_path={gen_protos.proto_dir}", "--include_imports", f"--descriptor_set_out={descriptor_path}", - str(workflowservice_request_response_proto), + *map(str, proto_files), ] ) -def generate_workflow_exports() -> None: - spec = spec_from_file_location( - "temporalio_nexus_system_workflow_service_exports", - output_dir / "__init__.py", - submodule_search_locations=[str(output_dir)], +def generate_package( + command: list[str], wit_name: str, output_dir: Path, *, native_api: bool +) -> None: + args = [*command, "python", str(wit_input_dir / wit_name), str(wit_deps_dir)] + if native_api: + args.append("--native-api") + subprocess.check_call( + [ + *args, + "--system-nexus", + "--support-file", + str(python_support_path), + "--descriptors", + str(output_dir.parent / "temporal_api.bin"), + "--output", + str(output_dir), + ] ) - if spec is None or spec.loader is None: - raise RuntimeError(f"Cannot load generated workflow service from {output_dir}") - module = module_from_spec(spec) - sys.modules[spec.name] = module - spec.loader.exec_module(module) - exports = cast(list[str], module.__all__) + +def merge_support_trees(support_dirs: list[Path], destination: Path) -> None: + merged_files: dict[Path, Path] = {} + for support_dir in support_dirs: + if not support_dir.is_dir(): + raise RuntimeError( + f"generator did not produce support directory: {support_dir}" + ) + for source in support_dir.rglob("*"): + if not source.is_file(): + continue + relative_path = source.relative_to(support_dir) + if previous := merged_files.get(relative_path): + if previous.read_bytes() != source.read_bytes(): + raise RuntimeError( + f"generated support files differ at {relative_path}: {previous} and {source}" + ) + else: + merged_files[relative_path] = source + destination.mkdir(parents=True, exist_ok=True) + for relative_path, source in merged_files.items(): + target = destination / relative_path + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, target) + + +def rewrite_support_imports(output_dir: Path) -> None: + for source in output_dir.rglob("*.py"): + if "_support" in source.relative_to(output_dir).parts: + continue + content = source.read_text() + rewritten = re.sub( + r"from \._support(\.[A-Za-z_][A-Za-z0-9_]*)? import ", + r"from temporalio.nexus._support\1 import ", + content, + ) + if rewritten != content: + source.write_text(rewritten) + + +def workflow_exports() -> list[str]: + tree = ast.parse((workflow_output_dir / "__init__.py").read_text()) + for statement in tree.body: + if not isinstance(statement, ast.Assign) or not any( + isinstance(target, ast.Name) and target.id == "__all__" + for target in statement.targets + ): + continue + value = ast.literal_eval(statement.value) + if not isinstance(value, list) or not all( + isinstance(item, str) for item in value + ): + raise RuntimeError( + "generated workflow package __all__ must be a list of strings" + ) + return value + raise RuntimeError("generated workflow package does not define __all__") + + +def replace_marker_block( + content: str, begin: str, end: str, replacement: list[str] +) -> str: + start = content.index(begin) + finish = content.index("\n", content.index(end, start)) + 1 + return content[:start] + "".join(replacement) + content[finish:] + + +def generate_workflow_exports() -> None: + exports = workflow_exports() import_block = [ "# BEGIN GENERATED NEXUS SYSTEM EXPORTS\n", "from temporalio.nexus.system.workflow_service import (\n", @@ -99,73 +184,91 @@ def generate_workflow_exports() -> None: " # END GENERATED NEXUS SYSTEM __ALL__\n", ] content = workflow_init_path.read_text() - start = content.index("# BEGIN GENERATED NEXUS SYSTEM EXPORTS") - end = content.index("# END GENERATED NEXUS SYSTEM EXPORTS", start) - end = content.index("\n", end) + 1 - content = content[:start] + "".join(import_block) + content[end:] - start = content.index(" # BEGIN GENERATED NEXUS SYSTEM __ALL__") - end = content.index(" # END GENERATED NEXUS SYSTEM __ALL__", start) - end = content.index("\n", end) + 1 - workflow_init_path.write_text(content[:start] + "".join(all_block) + content[end:]) + content = replace_marker_block( + content, import_block[0].strip(), import_block[-1].strip(), import_block + ) + workflow_init_path.write_text( + replace_marker_block( + content, + " # BEGIN GENERATED NEXUS SYSTEM __ALL__", + " # END GENERATED NEXUS SYSTEM __ALL__", + all_block, + ) + ) -def generate_nexus_system_api() -> None: - if not wit_path.exists(): - raise RuntimeError(f"missing WIT source: {wit_path}") - if not wit_deps_dir.exists(): - raise RuntimeError(f"missing WIT dependency directory: {wit_deps_dir}") - if not python_support_path.exists(): - raise RuntimeError(f"missing Python support source: {python_support_path}") +def publish_generated_packages( + staged_workflow: Path, staged_notification: Path +) -> None: + staged_support_dirs = [ + staged_workflow / "_support", + staged_notification / "_support", + ] + staged_support = staged_workflow.parent / "_support" + merge_support_trees(staged_support_dirs, staged_support) + for output_dir in (staged_workflow, staged_notification): + rewrite_support_imports(output_dir) + shutil.rmtree(output_dir / "_support") + for output_dir in ( + workflow_output_dir, + notification_output_dir, + support_output_dir, + ): + shutil.rmtree(output_dir, ignore_errors=True) + workflow_output_dir.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(staged_workflow, workflow_output_dir) + notification_output_dir.mkdir(parents=True) + shutil.copy2( + staged_notification / "models.py", notification_output_dir / "models.py" + ) + notification_output_dir.joinpath("__init__.py").write_text( + "from .models import OnCompleteRequest, OnCompleteRequestResult, OnCompleteResponse\n\n" + "__all__ = [\n" + ' "OnCompleteRequest",\n' + ' "OnCompleteRequestResult",\n' + ' "OnCompleteResponse",\n' + "]\n" + ) + shutil.copytree(staged_support, support_output_dir) + workflow_output_dir.parent.joinpath("__init__.py").touch() + +def generate_nexus_system_api() -> None: + required_paths = [ + wit_input_dir / "workflow-service.wit", + wit_input_dir / "notification-service.wit", + wit_deps_dir, + python_support_path, + *proto_files, + ] + for path in required_paths: + if not path.exists(): + raise RuntimeError(f"missing generator input: {path}") with tempfile.TemporaryDirectory(dir=base_dir) as temp_dir: - descriptor_path = Path(temp_dir) / "temporal_api.bin" + staging_dir = Path(temp_dir) + descriptor_path = staging_dir / "temporal_api.bin" build_descriptor_set(descriptor_path) command = nex_gen_command() - - shutil.rmtree(output_dir, ignore_errors=True) - output_dir.parent.mkdir(parents=True, exist_ok=True) - subprocess.check_call( - [ - *command, - "python", - str(wit_path), - str(wit_deps_dir), - "--native-api", - "--system-nexus", - "--support-file", - str(python_support_path), - "--descriptors", - str(descriptor_path), - "--output", - str(output_dir), - ] + staged_workflow = staging_dir / "workflow_service" + staged_notification = staging_dir / "notifications" + generate_package( + command, "workflow-service.wit", staged_workflow, native_api=True ) - - (output_dir.parent / "__init__.py").touch() + generate_package( + command, "notification-service.wit", staged_notification, native_api=False + ) + publish_generated_packages(staged_workflow, staged_notification) generate_workflow_exports() + format_paths = [ + str(workflow_output_dir), + str(notification_output_dir), + str(support_output_dir), + str(workflow_init_path), + ] subprocess.check_call( - [ - sys.executable, - "-m", - "ruff", - "check", - "--select", - "I", - "--fix", - str(output_dir), - str(workflow_init_path), - ] - ) - subprocess.check_call( - [ - sys.executable, - "-m", - "ruff", - "format", - str(output_dir), - str(workflow_init_path), - ] + [sys.executable, "-m", "ruff", "check", "--select", "I", "--fix", *format_paths] ) + subprocess.check_call([sys.executable, "-m", "ruff", "format", *format_paths]) if __name__ == "__main__": diff --git a/scripts/nex_gen_support.py b/scripts/nex_gen_support.py index fad21151e..bc4489e4e 100644 --- a/scripts/nex_gen_support.py +++ b/scripts/nex_gen_support.py @@ -6,6 +6,7 @@ import temporalio.api.common.v1.message_pb2 as common_pb2 import temporalio.api.enums.v1.workflow_pb2 as workflow_enums_pb2 +import temporalio.api.failure.v1.message_pb2 as failure_pb2 import temporalio.api.taskqueue.v1.message_pb2 as taskqueue_pb2 import temporalio.api.workflow.v1 import temporalio.common @@ -142,8 +143,11 @@ def _payload_to_value(payload: common_pb2.Payload) -> object: def payload_from_proto( proto: common_pb2.Payload, + type_hint: typing.Any = None, ) -> object: - return _payload_to_value(proto) + return temporalio.nexus.system._current_user_payload_converter().from_payloads( + [proto], [type_hint] if type_hint is not None else None + )[0] def payload_to_proto( @@ -152,6 +156,24 @@ def payload_to_proto( return _value_to_payload(payload) +def failure_from_proto( + proto: failure_pb2.Failure, +) -> BaseException: + return temporalio.nexus.system._current_user_failure_converter().from_failure( + proto, temporalio.nexus.system._current_user_payload_converter() + ) + + +def failure_to_proto( + failure: BaseException, +) -> failure_pb2.Failure: + proto = failure_pb2.Failure() + temporalio.nexus.system._current_user_failure_converter().to_failure( + failure, temporalio.nexus.system._current_user_payload_converter(), proto + ) + return proto + + def memo_from_proto( proto: common_pb2.Memo, ) -> collections.abc.Mapping[str, object]: diff --git a/temporalio/bridge/sdk-core b/temporalio/bridge/sdk-core index e163abd6d..70721e8a3 160000 --- a/temporalio/bridge/sdk-core +++ b/temporalio/bridge/sdk-core @@ -1 +1 @@ -Subproject commit e163abd6dc19040064986a63c8b8cf4756ffd320 +Subproject commit 70721e8a3c8d7f81203771d117c48eb1730acb2b diff --git a/temporalio/nexus/system/workflow_service/_support/__init__.py b/temporalio/nexus/_support/__init__.py similarity index 66% rename from temporalio/nexus/system/workflow_service/_support/__init__.py rename to temporalio/nexus/_support/__init__.py index 6af0c0c56..9c95e5949 100644 --- a/temporalio/nexus/system/workflow_service/_support/__init__.py +++ b/temporalio/nexus/_support/__init__.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py b/temporalio/nexus/_support/nex_gen_support.py similarity index 91% rename from temporalio/nexus/system/workflow_service/_support/nex_gen_support.py rename to temporalio/nexus/_support/nex_gen_support.py index fad21151e..bc4489e4e 100644 --- a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py +++ b/temporalio/nexus/_support/nex_gen_support.py @@ -6,6 +6,7 @@ import temporalio.api.common.v1.message_pb2 as common_pb2 import temporalio.api.enums.v1.workflow_pb2 as workflow_enums_pb2 +import temporalio.api.failure.v1.message_pb2 as failure_pb2 import temporalio.api.taskqueue.v1.message_pb2 as taskqueue_pb2 import temporalio.api.workflow.v1 import temporalio.common @@ -142,8 +143,11 @@ def _payload_to_value(payload: common_pb2.Payload) -> object: def payload_from_proto( proto: common_pb2.Payload, + type_hint: typing.Any = None, ) -> object: - return _payload_to_value(proto) + return temporalio.nexus.system._current_user_payload_converter().from_payloads( + [proto], [type_hint] if type_hint is not None else None + )[0] def payload_to_proto( @@ -152,6 +156,24 @@ def payload_to_proto( return _value_to_payload(payload) +def failure_from_proto( + proto: failure_pb2.Failure, +) -> BaseException: + return temporalio.nexus.system._current_user_failure_converter().from_failure( + proto, temporalio.nexus.system._current_user_payload_converter() + ) + + +def failure_to_proto( + failure: BaseException, +) -> failure_pb2.Failure: + proto = failure_pb2.Failure() + temporalio.nexus.system._current_user_failure_converter().to_failure( + failure, temporalio.nexus.system._current_user_payload_converter(), proto + ) + return proto + + def memo_from_proto( proto: common_pb2.Memo, ) -> collections.abc.Mapping[str, object]: diff --git a/temporalio/nexus/notifications/__init__.py b/temporalio/nexus/notifications/__init__.py new file mode 100644 index 000000000..72ec0651c --- /dev/null +++ b/temporalio/nexus/notifications/__init__.py @@ -0,0 +1,7 @@ +from .models import OnCompleteRequest, OnCompleteRequestResult, OnCompleteResponse + +__all__ = [ + "OnCompleteRequest", + "OnCompleteRequestResult", + "OnCompleteResponse", +] diff --git a/temporalio/nexus/notifications/models.py b/temporalio/nexus/notifications/models.py new file mode 100644 index 000000000..38ef42524 --- /dev/null +++ b/temporalio/nexus/notifications/models.py @@ -0,0 +1,149 @@ +# Generated by nexgen v0.2.7. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses +import typing + +import typing_extensions + +import temporalio.api.notificationservice.v1.request_response_pb2 +import temporalio.converter +from temporalio.nexus._support import ( + failure_from_proto, + failure_to_proto, + payload_from_proto, + payload_to_proto, +) + +OutputT = typing.TypeVar("OutputT") +SourceContextT = typing.TypeVar("SourceContextT") + + +@dataclasses.dataclass(slots=True) +class OnCompleteRequest(typing.Generic[OutputT, SourceContextT]): + result: OnCompleteRequestResult[OutputT] + source_context: SourceContextT + + +class _OnCompleteRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OnCompleteRequest[OutputT, SourceContextT], + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest, + ] +): + transfer_type: ( + type[ + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest + ] + | None + ) = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest, + type_hint: type[OnCompleteRequest[OutputT, SourceContextT]], + ) -> OnCompleteRequest[OutputT, SourceContextT]: + output_type, source_context_type = typing.get_args(type_hint) or ( + typing.Any, + typing.Any, + ) + _oneof_result: ( + OnCompleteRequestResultSuccess[OutputT] | OnCompleteRequestResultFailure + ) + match value.WhichOneof("result"): + case "success": + _oneof_result = OnCompleteRequestResultSuccess( + typing.cast(OutputT, payload_from_proto(value.success, output_type)) + ) + case "failure": + _oneof_result = OnCompleteRequestResultFailure( + failure_from_proto(value.failure) + ) + case None: + raise ValueError("missing required field OnCompleteRequest.result") + if not value.HasField("source_context"): + raise ValueError("missing required field OnCompleteRequest.source_context") + source_context = typing.cast( + SourceContextT, + payload_from_proto(value.source_context, source_context_type), + ) + return OnCompleteRequest( + result=_oneof_result, + source_context=source_context, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: OnCompleteRequest[OutputT, SourceContextT], + ) -> temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest: + message = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteRequest() + match value.result: + case OnCompleteRequestResultSuccess(): + message.success.CopyFrom(payload_to_proto(value.result.value)) + case OnCompleteRequestResultFailure(): + message.failure.CopyFrom(failure_to_proto(value.result.value)) + message.source_context.CopyFrom(payload_to_proto(value.source_context)) + return message + + +temporalio.converter.transfer_type_convertible( + _OnCompleteRequestTransferTypeConverter[typing.Any, typing.Any] +)(OnCompleteRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class OnCompleteResponse: + pass + + +class _OnCompleteResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + OnCompleteResponse, + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse, + ] +): + transfer_type: ( + type[ + temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse + ] + | None + ) = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse, + type_hint: type[OnCompleteResponse], + ) -> OnCompleteResponse: + return OnCompleteResponse() + + @typing_extensions.override + def to_transfer_type( + self, + value: OnCompleteResponse, + ) -> temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse: + message = temporalio.api.notificationservice.v1.request_response_pb2.OnCompleteResponse() + return message + + +temporalio.converter.transfer_type_convertible( + _OnCompleteResponseTransferTypeConverter +)(OnCompleteResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class OnCompleteRequestResultSuccess(typing.Generic[OutputT]): + value: OutputT + + +@dataclasses.dataclass(slots=True) +class OnCompleteRequestResultFailure: + value: BaseException + + +OnCompleteRequestResult = ( + OnCompleteRequestResultSuccess[OutputT] | OnCompleteRequestResultFailure +) diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index ed87c208f..8b9171b59 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations @@ -8,9 +8,9 @@ import nexusrpc import temporalio.converter +from temporalio.nexus._support import signal_with_start_workflow_serialization_context 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__ = [ diff --git a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py index c70b106db..85376e36a 100644 --- a/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py +++ b/temporalio/nexus/system/workflow_service/_system_nexus_interceptor.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index 232bc91ca..66dad150f 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations @@ -13,8 +13,7 @@ import temporalio.api.workflowservice.v1.request_response_pb2 import temporalio.common import temporalio.converter - -from ._support import ( +from temporalio.nexus._support import ( duration_from_proto, duration_to_proto, header_from_proto, @@ -232,9 +231,9 @@ def to_transfer_type( return message -_ = temporalio.converter.transfer_type_convertible( +temporalio.converter.transfer_type_convertible( _SignalWithStartWorkflowRequestTransferTypeConverter -)(SignalWithStartWorkflowRequest) +)(SignalWithStartWorkflowRequest) # pyright: ignore[reportUnusedCallResult] @dataclasses.dataclass(slots=True) @@ -285,9 +284,9 @@ def to_transfer_type( return message -_ = temporalio.converter.transfer_type_convertible(_UserMetadataTransferTypeConverter)( +temporalio.converter.transfer_type_convertible(_UserMetadataTransferTypeConverter)( UserMetadata -) +) # pyright: ignore[reportUnusedCallResult] @dataclasses.dataclass(slots=True) @@ -338,6 +337,6 @@ def to_transfer_type( return message -_ = temporalio.converter.transfer_type_convertible( +temporalio.converter.transfer_type_convertible( _SignalWithStartWorkflowResponseTransferTypeConverter -)(SignalWithStartWorkflowResponse) +)(SignalWithStartWorkflowResponse) # pyright: ignore[reportUnusedCallResult] diff --git a/temporalio/nexus/system/workflow_service/operations/__init__.py b/temporalio/nexus/system/workflow_service/operations/__init__.py index 28d71c338..07f1786b2 100644 --- a/temporalio/nexus/system/workflow_service/operations/__init__.py +++ b/temporalio/nexus/system/workflow_service/operations/__init__.py @@ -1,3 +1,3 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py index 2698b110a..2592507e7 100644 --- a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py +++ b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/services.py b/temporalio/nexus/system/workflow_service/services.py index c538493e1..e41afdc1e 100644 --- a/temporalio/nexus/system/workflow_service/services.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -1,4 +1,4 @@ -# Generated by nexgen v0.2.4. DO NOT EDIT! +# Generated by nexgen v0.2.7. DO NOT EDIT! from __future__ import annotations diff --git a/tests/nexus/test_notifications.py b/tests/nexus/test_notifications.py new file mode 100644 index 000000000..6f1f5cea2 --- /dev/null +++ b/tests/nexus/test_notifications.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import dataclasses + +import pytest + +import temporalio.converter +import temporalio.exceptions +import temporalio.nexus.notifications as notifications +import temporalio.nexus.notifications.models as models +import temporalio.nexus.system as nexus_system + + +@dataclasses.dataclass +class NotificationValue: + message: str + + +@pytest.mark.parametrize("success", [True, False]) +def test_notification_request_roundtrip(success: bool) -> None: + data_converter = temporalio.converter.default() + converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + result: models.OnCompleteRequestResult[NotificationValue] = ( + models.OnCompleteRequestResultSuccess(NotificationValue("result")) + if success + else models.OnCompleteRequestResultFailure(ValueError("failure")) + ) + request = notifications.OnCompleteRequest( + result=result, source_context=NotificationValue("context") + ) + payload = converter.to_payload(request) + decoded = converter.from_payload( + payload, + notifications.OnCompleteRequest[NotificationValue, NotificationValue], + ) + assert decoded.source_context == request.source_context + assert isinstance(decoded.source_context, NotificationValue) + if success: + assert isinstance(decoded.result, models.OnCompleteRequestResultSuccess) + assert decoded.result.value == NotificationValue("result") + assert isinstance(decoded.result.value, NotificationValue) + else: + assert isinstance(decoded.result, models.OnCompleteRequestResultFailure) + assert isinstance(decoded.result.value, temporalio.exceptions.ApplicationError) + assert decoded.result.value.message == "failure" + assert decoded.result.value.type == "ValueError" + + +def test_notification_response_roundtrip() -> None: + data_converter = temporalio.converter.default() + converter = nexus_system._get_payload_converter( + data_converter.payload_converter, + data_converter.failure_converter, + ) + response = notifications.OnCompleteResponse() + assert ( + converter.from_payload( + converter.to_payload(response), notifications.OnCompleteResponse + ) + == response + )