diff --git a/GUIDE.md b/GUIDE.md index 23fb06849..b923bdff3 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -29,6 +29,7 @@ of `@nexus` directives. - [Omitted Fields](#omitted-fields) - [Flattened Records](#flattened-records) - [Output Transforms](#output-transforms) +- [Callers Outside a Workflow](#callers-outside-a-workflow) - [Function References](#function-references) - [Directive Glossary](#directive-glossary) @@ -1056,6 +1057,46 @@ return handles.FromSignalWithStart(request, &result) --- +## Callers Outside a Workflow + +With `--client`, the Python target also writes `client.py` beside `models.py`. +It declares one caller per service, `Client`, for code that is not in a +Workflow, such as an Activity or a client process. The caller does not choose +how a call travels. It takes a `Transport`, an object with one method: + +```python +class Transport(typing.Protocol): + async def call( + self, + *, + service: str, + operation: str, + input: temporalio.api.common.v1.Payload, + ) -> temporalio.api.common.v1.Payload: ... +``` + +Each caller method converts its request to a payload, gives it to the transport, +and converts the answer. A proto-backed model travels as a `binary/protobuf` +payload of its proto message, so a transport that moves proto bytes needs +nothing else. An answer for it can be `binary/protobuf`, `json/protobuf` or +`json/plain`. The caller takes a `data_converter`, the SDK default when none is +given. Each payload and failure inside a proto-backed model goes through its +converters and codec, as in Workflow code. The model's own payload doesn't, +since the service reads it. A model that is not proto-backed is one user value, +and the data converter converts and encodes it whole. A transport raises +`TransportError` for a call that failed, with `retryable` set when the same call +can succeed later. + +```python +client = StartWorkflowServiceClient(my_transport) +response = await client.cancel_workflow(request) +``` + +The caller supports operations whose input and output are models of the same +module. An operation whose models take type parameters, or that returns a +resource without a wire type, is a generation error. TypeScript and .NET do not +generate a caller yet. + ## Function References Nexus and Temporal often need to identify a function by name on the wire, while diff --git a/advanced/README.md b/advanced/README.md index d35ba41a3..e787b9237 100644 --- a/advanced/README.md +++ b/advanced/README.md @@ -182,8 +182,11 @@ cargo run --features advanced -- dotnet \ Inputs are positional; pass more than one path (a file or a directory) to link additional WIT into the parser workspace. Generation produces definitions by -default. Add `--native-api` to also generate native API bindings. Add `--format` -to run a formatter after generation: +default. Add `--native-api` to also generate native API bindings. Add `--client` +to also generate, in Python, a `client.py` with one caller per service for code +outside a Workflow. The caller sends each call through a `Transport` that you +give, so the same caller works over HTTP, a bridge, or a test double. Add +`--format` to run a formatter after generation: - Python: `ruff format` - TypeScript: `prettier --write` diff --git a/advanced/samples/inputs/examples.toml b/advanced/samples/inputs/examples.toml index 042784ac2..963ed4376 100644 --- a/advanced/samples/inputs/examples.toml +++ b/advanced/samples/inputs/examples.toml @@ -12,3 +12,6 @@ languages = ["python", "dotnet", "typescript"] [notification-service] languages = ["python", "dotnet", "typescript"] + +[start-workflow] +client = true diff --git a/advanced/samples/python/tests/test_start_workflow.py b/advanced/samples/python/tests/test_start_workflow.py index d48abf66a..59f9fb365 100644 --- a/advanced/samples/python/tests/test_start_workflow.py +++ b/advanced/samples/python/tests/test_start_workflow.py @@ -206,3 +206,38 @@ async def test_start_workflow_returns_wrapper_handle( result_error.value.args[0] == "started-workflow.get_result is not yet implemented" ) + + +async def test_client_calls_an_operation_over_a_transport() -> None: + from temporalio.api.common.v1 import Payload + from temporalio.api.workflowservice.v1 import ( + RequestCancelWorkflowExecutionRequest, + RequestCancelWorkflowExecutionResponse, + ) + import wit.start_workflow.client as start_workflow_client + + sent: list[RequestCancelWorkflowExecutionRequest] = [] + + class Transport: + async def call( + self, *, service: str, operation: str, input: Payload + ) -> Payload: + assert (service, operation) == ("StartWorkflowService", "CancelWorkflow") + sent.append(RequestCancelWorkflowExecutionRequest.FromString(input.data)) + answer = RequestCancelWorkflowExecutionResponse() + return Payload( + metadata={"encoding": b"binary/protobuf"}, + data=answer.SerializeToString(), + ) + + client = start_workflow_client.StartWorkflowServiceClient(Transport()) + request = start_workflow_models.CancelWorkflowRequest( + namespace="default", + workflow_execution=start_workflow_models.WorkflowExecution(workflow_id="wf"), + reason="done", + ) + response = await client.cancel_workflow(request) + + assert isinstance(response, start_workflow_models.CancelWorkflowResponse) + assert sent[0].workflow_execution.workflow_id == "wf" + assert sent[0].reason == "done" diff --git a/advanced/samples/python/wit/start_workflow/client.py b/advanced/samples/python/wit/start_workflow/client.py new file mode 100644 index 000000000..327401427 --- /dev/null +++ b/advanced/samples/python/wit/start_workflow/client.py @@ -0,0 +1,230 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import contextlib +import typing + +import google.protobuf.json_format +import google.protobuf.message +import temporalio.api.common.v1 +import temporalio.converter +import temporalio.nexus.system + +from .models import ( + CancelWorkflowRequest, + CancelWorkflowResponse, + StartWorkflowRequest, + StartWorkflowResult, + _CancelWorkflowRequestTransferTypeConverter, + _CancelWorkflowResponseTransferTypeConverter, + _StartWorkflowRequestTransferTypeConverter, + _StartWorkflowResultTransferTypeConverter, +) + + +class Transport(typing.Protocol): + """Carries one operation call to a service and its answer back. + + A transport moves payloads only. The caller converts its models to and + from payloads. A proto-backed model travels as a ``binary/protobuf`` + payload, and an answer for one can be ``binary/protobuf``, + ``json/protobuf`` or ``json/plain``. + """ + + async def call( + self, + *, + service: str, + operation: str, + input: temporalio.api.common.v1.Payload, + ) -> temporalio.api.common.v1.Payload: + """Send one call and return its answer. + + Raise :py:class:`TransportError` for a call that failed. + """ + ... + + +class TransportError(Exception): + """A call that the transport could not complete. + + ``retryable`` says whether the same call can succeed later. + """ + + def __init__(self, message: str, *, retryable: bool = False) -> None: + """Create a transport error.""" + super().__init__(message) + self.retryable: bool = retryable + + +_MessageT = typing.TypeVar("_MessageT", bound=google.protobuf.message.Message) + + +def _proto_payload( + message: google.protobuf.message.Message, +) -> temporalio.api.common.v1.Payload: + return temporalio.api.common.v1.Payload( + metadata={ + "encoding": b"binary/protobuf", + "messageType": message.DESCRIPTOR.full_name.encode(), + }, + data=message.SerializeToString(), + ) + + +def _proto_message( + payload: temporalio.api.common.v1.Payload, + message_type: type[_MessageT] | None, +) -> _MessageT: + if message_type is None: + raise TypeError("the answer model has no proto message type") + message = message_type() + encoding = payload.metadata.get("encoding", b"") + if encoding == b"binary/protobuf": + _ = message.ParseFromString(payload.data) + elif encoding in (b"json/protobuf", b"json/plain"): + _ = google.protobuf.json_format.Parse(payload.data, message) + elif encoding != b"binary/null": + raise ValueError( + f"cannot read {message.DESCRIPTOR.full_name} from a {encoding.decode()!r} payload" + ) + return message + + +# The codec runs on the user payloads inside a proto-backed model, failure details +# included, and never on the model's own payload, which the service reads. +def _payloads_in( + message: google.protobuf.message.Message, +) -> list[temporalio.api.common.v1.Payload]: + if isinstance(message, temporalio.api.common.v1.Payload): + return [message] + found: list[temporalio.api.common.v1.Payload] = [] + for _, value in message.ListFields(): + if isinstance(value, google.protobuf.message.Message): + found.extend(_payloads_in(value)) + elif not isinstance(value, (str, bytes, int, float, bool)): + items = typing.cast( + "collections.abc.Iterable[object]", + value.values() if hasattr(value, "values") else value, + ) + for item in items: + if isinstance(item, google.protobuf.message.Message): + found.extend(_payloads_in(item)) + return found + + +async def _encode_payloads( + message: google.protobuf.message.Message, + codec: temporalio.converter.PayloadCodec | None, +) -> None: + if codec is None: + return + for payload in _payloads_in(message): + [encoded] = await codec.encode([payload]) + payload.CopyFrom(encoded) + + +async def _decode_payloads( + message: google.protobuf.message.Message, + codec: temporalio.converter.PayloadCodec | None, +) -> None: + if codec is None: + return + for payload in _payloads_in(message): + [decoded] = await codec.decode([payload]) + payload.CopyFrom(decoded) + + +class StartWorkflowServiceClient: + """Calls the `StartWorkflowService` operations over a transport. + + For code outside a Workflow. Workflow code uses the operation stubs. + """ + + def __init__( + self, + transport: Transport, + *, + data_converter: temporalio.converter.DataConverter | None = None, + ) -> None: + """Create a caller. + + Args: + transport: Carries each call to the service. + data_converter: Converts and encodes the user values. A proto-backed + model travels as is, and each payload and failure inside it goes + through the converter and the codec, as in Workflow code. A model + that is not proto-backed is one user value. Defaults to the SDK's + default data converter. + """ + self._transport: Transport = transport + self._data_converter: temporalio.converter.DataConverter = ( + data_converter or temporalio.converter.DataConverter.default + ) + + def _converters(self) -> contextlib.AbstractContextManager[None]: + return temporalio.nexus.system._user_converter_context( + temporalio.nexus.system._SystemNexusUserConverters( + payload_converter=self._data_converter.payload_converter, + failure_converter=self._data_converter.failure_converter, + ) + ) + + async def _call( + self, operation: str, payload: temporalio.api.common.v1.Payload + ) -> temporalio.api.common.v1.Payload: + return await self._transport.call( + service="StartWorkflowService", + operation=operation, + input=payload, + ) + + async def start_workflow( + self, request: StartWorkflowRequest + ) -> StartWorkflowResult: + with self._converters(): + message = _StartWorkflowRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("StartWorkflow", payload) + converter = _StartWorkflowResultTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + return converter.from_transfer_type(message, StartWorkflowResult) + + async def restart_workflow( + self, request: StartWorkflowRequest + ) -> StartWorkflowResult: + with self._converters(): + message = _StartWorkflowRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("RestartWorkflow", payload) + converter = _StartWorkflowResultTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + return converter.from_transfer_type(message, StartWorkflowResult) + + async def cancel_workflow( + self, request: CancelWorkflowRequest + ) -> CancelWorkflowResponse: + with self._converters(): + message = _CancelWorkflowRequestTransferTypeConverter().to_transfer_type( + request + ) + await _encode_payloads(message, self._data_converter.payload_codec) + payload = _proto_payload(message) + answer = await self._call("CancelWorkflow", payload) + converter = _CancelWorkflowResponseTransferTypeConverter() + message = _proto_message(answer, converter.transfer_type) + await _decode_payloads(message, self._data_converter.payload_codec) + with self._converters(): + return converter.from_transfer_type(message, CancelWorkflowResponse) diff --git a/changelog/added/relay-raccoon.md b/changelog/added/relay-raccoon.md new file mode 100644 index 000000000..270be946e --- /dev/null +++ b/changelog/added/relay-raccoon.md @@ -0,0 +1 @@ +Python generation accepts `--client`, which writes `client.py` with one caller per service over a `Transport` that the caller gives. diff --git a/src/error.rs b/src/error.rs index 78995c800..62a99d9f2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -215,6 +215,14 @@ pub enum Error { )] IncompleteOperationOutputTransform { service: String, operation: String }, + #[error("{language} cannot generate a caller for operation `{service}.{operation}`: {reason}")] + UnsupportedClientOperation { + language: Language, + service: String, + operation: String, + reason: String, + }, + #[error( "service `{service}` operation `{operation}` references unknown input proto type `{type_name}`" )] diff --git a/src/generator/go.rs b/src/generator/go.rs index 8adc5d7b0..d7596caec 100644 --- a/src/generator/go.rs +++ b/src/generator/go.rs @@ -5838,6 +5838,7 @@ mod tests { config: crate::nexgen_config::NexgenConfig { mode: crate::generator::GenerationMode::NativeApi, system_nexus, + client: false, }, language: crate::language::Language::Go, input_paths: vec![input], diff --git a/src/generator/python.rs b/src/generator/python.rs index 4d73f7361..2585e5816 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -700,6 +700,13 @@ impl<'a> ApiPlanner<'a> { GeneratedFileOrigin::fixed("generated Python resources package initializer"), )?; } + if crate::nexgen_config::current().client && !services.is_empty() { + files.insert( + "client.py", + render_client_module(services)?, + module_origin.clone(), + )?; + } if !services.is_empty() { files.insert( "services.py", @@ -1313,9 +1320,23 @@ impl<'a> ApiPlanner<'a> { .map(|usage| usage.parameter.name) .collect(), generic_model_types, + client_input_proto_model: input.and_then(|input| self.proto_model_name(input)), + client_output_proto_model: operation + .output_type() + .and_then(|output| self.proto_model_name(output)), }) } + fn proto_model_name(&self, model_type: &PlannedType) -> Option { + let PlannedType::Record(record) = model_type else { + return None; + }; + self.api_plan + .record(&record.full_name) + .filter(|record| record.data.proto.is_some()) + .map(|record| record.name.clone()) + } + fn resolve_output_annotation(&mut self, model_type: &PlannedType) -> Result { match model_type { PlannedType::External(ExternalTypeSpec::Proto(_)) @@ -2546,6 +2567,11 @@ struct RenderedOperation<'a> { unpacked_input: Option, model_input_type_parameters: Vec, generic_model_types: bool, + /// The generated proto-backed model of the input and of the output, when + /// each is one. A caller outside a Workflow converts these through their + /// transfer-type converters instead of the payload converter. + client_input_proto_model: Option, + client_output_proto_model: Option, } #[derive(Debug)] @@ -3992,6 +4018,392 @@ fn render_models_module( Ok(Some(output)) } +/// The model that a caller outside a Workflow uses for an operation's input or +/// output. The caller supports the models of this module only. +fn client_model_name( + service: &RenderedService<'_>, + operation: &RenderedOperation<'_>, + side: &str, + type_ref: &str, + module_path: Option<&str>, + generic: bool, +) -> Result { + let unsupported = |reason: String| Error::UnsupportedClientOperation { + language: Language::Python, + service: service.name.to_string(), + operation: operation.name.to_string(), + reason, + }; + if generic { + return Err(unsupported("its models take type parameters".to_string())); + } + if module_path != Some(".models") { + return Err(unsupported(format!( + "its {side} `{type_ref}` is not a model of this module" + ))); + } + Ok(type_ref + .strip_prefix("models.") + .unwrap_or(type_ref) + .to_string()) +} + +/// One caller per service for code outside a Workflow. Each method converts its +/// model to a payload, hands the payload to the transport, and converts the +/// answer back. A proto-backed model converts through its transfer-type +/// converter, so the transport moves proto bytes and needs no message type. +fn render_client_module(services: &[RenderedService<'_>]) -> Result { + let mut model_names = BTreeSet::new(); + let mut body = String::new(); + body.push_str(CLIENT_RUNTIME); + for service in services { + body.push_str("\n\n"); + body.push_str("class "); + body.push_str(&client_class_name(service)); + body.push_str(":\n"); + body.push_str(" \"\"\"Calls the "); + body.push_str(&python_docstring_literal_text(&format!( + "`{}`", + service.wire_name + ))); + body.push_str(" operations over a transport.\n\n"); + body.push_str(" For code outside a Workflow. Workflow code uses the operation stubs.\n"); + body.push_str(" \"\"\"\n\n"); + body.push_str(" def __init__(\n"); + body.push_str(" self,\n"); + body.push_str(" transport: Transport,\n"); + body.push_str(" *,\n"); + body.push_str( + " data_converter: temporalio.converter.DataConverter | None = None,\n", + ); + body.push_str(" ) -> None:\n"); + body.push_str(" \"\"\"Create a caller.\n\n"); + body.push_str(" Args:\n"); + body.push_str(" transport: Carries each call to the service.\n"); + body.push_str( + " data_converter: Converts and encodes the user values. A proto-backed\n", + ); + body.push_str( + " model travels as is, and each payload and failure inside it goes\n", + ); + body.push_str( + " through the converter and the codec, as in Workflow code. A model\n", + ); + body.push_str( + " that is not proto-backed is one user value. Defaults to the SDK's\n", + ); + body.push_str(" default data converter.\n"); + body.push_str(" \"\"\"\n"); + body.push_str(" self._transport: Transport = transport\n"); + body.push_str(" self._data_converter: temporalio.converter.DataConverter = (\n"); + body.push_str(" data_converter or temporalio.converter.DataConverter.default\n"); + body.push_str(" )\n"); + body.push_str("\n"); + body.push_str(" def _converters(self) -> contextlib.AbstractContextManager[None]:\n"); + body.push_str(" return temporalio.nexus.system._user_converter_context(\n"); + body.push_str(" temporalio.nexus.system._SystemNexusUserConverters(\n"); + body.push_str( + " payload_converter=self._data_converter.payload_converter,\n", + ); + body.push_str( + " failure_converter=self._data_converter.failure_converter,\n", + ); + body.push_str(" )\n"); + body.push_str(" )\n"); + body.push_str("\n"); + body.push_str(" async def _call(\n"); + body.push_str(" self, operation: str, payload: temporalio.api.common.v1.Payload\n"); + body.push_str(" ) -> temporalio.api.common.v1.Payload:\n"); + body.push_str(" return await self._transport.call(\n"); + body.push_str(" service="); + body.push_str(&python_string_literal(service.wire_name)); + body.push_str(",\n"); + body.push_str(" operation=operation,\n"); + body.push_str(" input=payload,\n"); + body.push_str(" )\n"); + for operation in &service.operations { + let Some(input) = &operation.input else { + return Err(Error::UnsupportedClientOperation { + language: Language::Python, + service: service.name.to_string(), + operation: operation.name.to_string(), + reason: "it has no input".to_string(), + }); + }; + // A resource with a wire type answers with its wire model, and the + // caller returns that model. A resource without one has no model. + if operation.output_module_path.as_deref() == Some("._resources") { + return Err(Error::UnsupportedClientOperation { + language: Language::Python, + service: service.name.to_string(), + operation: operation.name.to_string(), + reason: "it returns a resource".to_string(), + }); + } + let input_model = client_model_name( + service, + operation, + "input", + &input.type_ref, + input.module_path.as_deref(), + operation.generic_model_types, + )?; + model_names.insert(input_model.clone()); + let output_model = if operation.output_none { + None + } else { + let output_model = client_model_name( + service, + operation, + "output", + &operation.output_ref, + operation.output_module_path.as_deref(), + operation.generic_model_types, + )?; + model_names.insert(output_model.clone()); + Some(output_model) + }; + for proto_model in operation + .client_input_proto_model + .iter() + .chain(operation.client_output_proto_model.iter()) + { + model_names.insert(format!("_{proto_model}TransferTypeConverter")); + } + render_client_method(&mut body, operation, &input_model, output_model.as_deref()); + } + } + + // A helper that no method uses would be dead code in the module. + let uses_proto = services + .iter() + .flat_map(|service| service.operations.iter()) + .any(|operation| { + operation.client_input_proto_model.is_some() + || operation.client_output_proto_model.is_some() + }); + if uses_proto { + body.insert_str(CLIENT_RUNTIME.len(), CLIENT_PROTO_RUNTIME); + } + + let mut output = String::new(); + render_generated_file_header(&mut output); + output.push('\n'); + if uses_proto { + output.push_str("import collections.abc\n"); + } + output.push_str("import contextlib\n"); + output.push_str("import typing\n\n"); + if uses_proto { + output.push_str("import google.protobuf.json_format\n"); + output.push_str("import google.protobuf.message\n"); + } + output.push_str("import temporalio.api.common.v1\n"); + output.push_str("import temporalio.converter\n"); + output.push_str("import temporalio.nexus.system\n"); + output.push('\n'); + render_named_python_import( + &mut output, + ".models", + &model_names.into_iter().collect::>(), + ); + output.push_str("\n\n"); + output.push_str(&body); + Ok(output) +} + +fn client_class_name(service: &RenderedService<'_>) -> String { + format!("{}Client", service.name) +} + +fn render_client_method( + output: &mut String, + operation: &RenderedOperation<'_>, + input_model: &str, + output_model: Option<&str>, +) { + output.push('\n'); + output.push_str(" async def "); + output.push_str(&operation.attr_name); + output.push_str("(self, request: "); + output.push_str(input_model); + output.push_str(") -> "); + output.push_str(output_model.unwrap_or("None")); + output.push_str(":\n"); + render_python_docstring( + output, + " ", + operation.doc.as_deref(), + &[], + None, + operation.experimental, + ); + let codec = "self._data_converter.payload_codec"; + match &operation.client_input_proto_model { + Some(proto_model) => { + output.push_str(" with self._converters():\n"); + output.push_str(" message = _"); + output.push_str(proto_model); + output.push_str("TransferTypeConverter().to_transfer_type(request)\n"); + output.push_str(" await _encode_payloads(message, "); + output.push_str(codec); + output.push_str(")\n"); + output.push_str(" payload = _proto_payload(message)\n"); + } + None => { + output.push_str(" [payload] = await self._data_converter.encode([request])\n"); + } + } + let Some(output_model) = output_model else { + output.push_str(" _ = await self._call("); + output.push_str(&python_string_literal(operation.wire_name)); + output.push_str(", payload)\n"); + return; + }; + output.push_str(" answer = await self._call("); + output.push_str(&python_string_literal(operation.wire_name)); + output.push_str(", payload)\n"); + match &operation.client_output_proto_model { + Some(proto_model) => { + output.push_str(" converter = _"); + output.push_str(proto_model); + output.push_str("TransferTypeConverter()\n"); + output.push_str(" message = _proto_message(answer, converter.transfer_type)\n"); + output.push_str(" await _decode_payloads(message, "); + output.push_str(codec); + output.push_str(")\n"); + output.push_str(" with self._converters():\n"); + output.push_str(" return converter.from_transfer_type(message, "); + output.push_str(output_model); + output.push_str(")\n"); + } + None => { + output.push_str(" [value] = await self._data_converter.decode([answer], ["); + output.push_str(output_model); + output.push_str("])\n"); + output.push_str(" return typing.cast("); + output.push_str(output_model); + output.push_str(", value)\n"); + } + } +} + +const CLIENT_RUNTIME: &str = r#"class Transport(typing.Protocol): + """Carries one operation call to a service and its answer back. + + A transport moves payloads only. The caller converts its models to and + from payloads. A proto-backed model travels as a ``binary/protobuf`` + payload, and an answer for one can be ``binary/protobuf``, + ``json/protobuf`` or ``json/plain``. + """ + + async def call( + self, + *, + service: str, + operation: str, + input: temporalio.api.common.v1.Payload, + ) -> temporalio.api.common.v1.Payload: + """Send one call and return its answer. + + Raise :py:class:`TransportError` for a call that failed. + """ + ... + + +class TransportError(Exception): + """A call that the transport could not complete. + + ``retryable`` says whether the same call can succeed later. + """ + + def __init__(self, message: str, *, retryable: bool = False) -> None: + """Create a transport error.""" + super().__init__(message) + self.retryable: bool = retryable +"#; + +/// The helpers of a caller that moves proto-backed models. +const CLIENT_PROTO_RUNTIME: &str = r#" + +_MessageT = typing.TypeVar("_MessageT", bound=google.protobuf.message.Message) + + +def _proto_payload( + message: google.protobuf.message.Message, +) -> temporalio.api.common.v1.Payload: + return temporalio.api.common.v1.Payload( + metadata={ + "encoding": b"binary/protobuf", + "messageType": message.DESCRIPTOR.full_name.encode(), + }, + data=message.SerializeToString(), + ) + + +def _proto_message( + payload: temporalio.api.common.v1.Payload, + message_type: type[_MessageT] | None, +) -> _MessageT: + if message_type is None: + raise TypeError("the answer model has no proto message type") + message = message_type() + encoding = payload.metadata.get("encoding", b"") + if encoding == b"binary/protobuf": + _ = message.ParseFromString(payload.data) + elif encoding in (b"json/protobuf", b"json/plain"): + _ = google.protobuf.json_format.Parse(payload.data, message) + elif encoding != b"binary/null": + raise ValueError( + f"cannot read {message.DESCRIPTOR.full_name} from a {encoding.decode()!r} payload" + ) + return message + + +# The codec runs on the user payloads inside a proto-backed model, failure details +# included, and never on the model's own payload, which the service reads. +def _payloads_in( + message: google.protobuf.message.Message, +) -> list[temporalio.api.common.v1.Payload]: + if isinstance(message, temporalio.api.common.v1.Payload): + return [message] + found: list[temporalio.api.common.v1.Payload] = [] + for _, value in message.ListFields(): + if isinstance(value, google.protobuf.message.Message): + found.extend(_payloads_in(value)) + elif not isinstance(value, (str, bytes, int, float, bool)): + items = typing.cast( + "collections.abc.Iterable[object]", + value.values() if hasattr(value, "values") else value, + ) + for item in items: + if isinstance(item, google.protobuf.message.Message): + found.extend(_payloads_in(item)) + return found + + +async def _encode_payloads( + message: google.protobuf.message.Message, + codec: temporalio.converter.PayloadCodec | None, +) -> None: + if codec is None: + return + for payload in _payloads_in(message): + [encoded] = await codec.encode([payload]) + payload.CopyFrom(encoded) + + +async def _decode_payloads( + message: google.protobuf.message.Message, + codec: temporalio.converter.PayloadCodec | None, +) -> None: + if codec is None: + return + for payload in _payloads_in(message): + [decoded] = await codec.decode([payload]) + payload.CopyFrom(decoded) +"#; + fn render_resources_package_init(services: &[RenderedService<'_>]) -> String { let resources = services .iter() diff --git a/src/main.rs b/src/main.rs index 0afa1763c..ca8f17c0e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -68,6 +68,10 @@ struct GenerateArgs { #[cfg(feature = "advanced")] #[arg(long = "system-nexus")] system_nexus: bool, + /// Generate a caller per service for code outside a Workflow (Python). + #[cfg(feature = "advanced")] + #[arg(long = "client")] + client: bool, } #[derive(Args)] @@ -188,6 +192,7 @@ fn main() -> ExitCode { #[cfg(feature = "advanced")] { let system_nexus = args.common.system_nexus; + let client = args.common.client; let config = NexgenConfig { mode: if args.common.generate_native_api { nexgen::generator::GenerationMode::NativeApi @@ -195,6 +200,7 @@ fn main() -> ExitCode { nexgen::generator::GenerationMode::DefinitionsOnly }, system_nexus, + client, }; let mut request = generate_request( Language::Python, @@ -221,6 +227,7 @@ fn main() -> ExitCode { #[cfg(feature = "advanced")] { let system_nexus = args.common.system_nexus; + let client = args.common.client; let config = NexgenConfig { mode: if args.common.generate_native_api { nexgen::generator::GenerationMode::NativeApi @@ -228,6 +235,7 @@ fn main() -> ExitCode { nexgen::generator::GenerationMode::DefinitionsOnly }, system_nexus, + client, }; let mut request = generate_request( Language::TypeScript, @@ -294,6 +302,7 @@ fn generate_request( nexgen::generator::GenerationMode::DefinitionsOnly }, system_nexus: args.system_nexus, + client: args.client, }, #[cfg(not(feature = "advanced"))] config: Default::default(), diff --git a/src/nexgen_config.rs b/src/nexgen_config.rs index 28038e06c..65abe315c 100644 --- a/src/nexgen_config.rs +++ b/src/nexgen_config.rs @@ -11,6 +11,9 @@ use crate::generator::GenerationMode; pub struct NexgenConfig { pub mode: GenerationMode, pub system_nexus: bool, + /// Also generate a caller per service for code outside a Workflow, over a + /// transport that the caller gives. + pub client: bool, } impl Default for NexgenConfig { @@ -18,6 +21,7 @@ impl Default for NexgenConfig { Self { mode: GenerationMode::DefinitionsOnly, system_nexus: false, + client: false, } } } @@ -58,6 +62,7 @@ mod tests { let outer = NexgenConfig { mode: GenerationMode::DefinitionsOnly, system_nexus: true, + client: false, }; { let _outer_scope = scope(outer); @@ -77,6 +82,7 @@ mod tests { let config = NexgenConfig { mode: GenerationMode::DefinitionsOnly, system_nexus: true, + client: false, }; { let _scope = scope(config); diff --git a/tests/generate_go.rs b/tests/generate_go.rs index fabc8aa20..95818efd6 100644 --- a/tests/generate_go.rs +++ b/tests/generate_go.rs @@ -66,6 +66,7 @@ fn generate_to_string_with_system_nexus( config: nexgen::nexgen_config::NexgenConfig { mode: nexgen::generator::GenerationMode::NativeApi, system_nexus, + client: false, }, language, input_paths: input_paths.to_vec(), @@ -1466,6 +1467,7 @@ interface sample-service {{ config: nexgen::nexgen_config::NexgenConfig { mode: nexgen::generator::GenerationMode::NativeApi, system_nexus, + client: false, }, language: nexgen::language::Language::Go, input_paths: vec![wit_path], diff --git a/tests/wit_streaming.rs b/tests/wit_streaming.rs index 1ed3412ec..fcd6e6589 100644 --- a/tests/wit_streaming.rs +++ b/tests/wit_streaming.rs @@ -1,5 +1,6 @@ -//! WIT annotations for streaming services: `@nexus.offset` on record fields -//! and `@nexus.long-poll` on operations. +//! WIT annotations for streaming services (`@nexus.offset` on record fields +//! and `@nexus.long-poll` on operations) and the caller that `--client` +//! generates for code outside a Workflow. #![cfg(feature = "advanced")] use std::collections::BTreeMap; @@ -122,22 +123,32 @@ fn try_generate( language: Language, mode: GenerationMode, wit: &str, +) -> Result, String> { + let config = NexgenConfig { + mode, + ..Default::default() + }; + try_generate_with(config, language, wit, Vec::new()) +} + +fn try_generate_with( + config: NexgenConfig, + language: Language, + wit: &str, + extra_inputs: Vec, ) -> Result, String> { let temp_dir = unique_output_path(&format!("{language:?}")); fs::create_dir_all(&temp_dir).unwrap(); let input_path = temp_dir.join("logs.wit"); fs::write(&input_path, wit).unwrap(); let output_path = temp_dir.join("generated"); + let mut input_paths = vec![input_path]; + input_paths.extend(extra_inputs); + input_paths.push(project_root().join("advanced/samples/inputs/deps")); let result = generate_to_file(&GenerateRequest { - config: NexgenConfig { - mode, - ..Default::default() - }, + config, language, - input_paths: vec![ - input_path, - project_root().join("advanced/samples/inputs/deps"), - ], + input_paths, support_package: Some( match language { Language::Dotnet => "Acme.Support", @@ -203,14 +214,6 @@ fn python_declares_one_offset_type_for_every_field_that_names_it() { ); } -#[test] -fn offset_fields_generate_in_typescript_and_dotnet() { - for language in [Language::TypeScript, Language::Dotnet] { - let files = generate(language, GenerationMode::NativeApi, LOG_WIT); - assert!(!files.is_empty(), "{language:?}"); - } -} - #[test] fn offset_directive_rejects_a_field_that_is_not_an_integer() { let error = generate_error( @@ -226,54 +229,6 @@ fn offset_directive_rejects_a_field_that_is_not_an_integer() { ); } -#[test] -fn offset_directive_requires_a_kebab_case_type_name() { - for (directive, reason) in [ - ("@nexus.offset", "missing the offset type name"), - ( - "@nexus.offset \"LogOffset\"", - "not a kebab-case WIT identifier", - ), - ( - "@nexus.offset \"log-offset\" kind=\"cursor\"", - "takes only the offset type name", - ), - ] { - let error = generate_error( - Language::Python, - &LOG_WIT.replacen("@nexus.offset \"log-offset\"", directive, 1), - ); - assert!(error.contains(reason), "{directive}: {error}"); - } -} - -#[test] -fn offset_directive_cannot_be_combined_with_omit() { - let error = generate_error( - Language::Python, - &LOG_WIT.replace( - " /// @nexus.offset \"log-offset\"\n offset: option,", - " /// @nexus.omit\n /// @nexus.offset \"log-offset\"\n offset: option,", - ), - ); - assert!( - error.contains("cannot be combined with `@nexus.offset`"), - "{error}" - ); -} - -#[test] -fn python_offset_type_cannot_share_a_name_with_a_model() { - let error = generate_error( - Language::Python, - &LOG_WIT.replace("\"log-offset\"", "\"entry\""), - ); - assert!( - error.contains("`Entry`") && error.contains("offset type"), - "{error}" - ); -} - /// Replaces the long-poll directive of `read` with `directive`. fn with_long_poll(directive: &str) -> String { LOG_WIT.replace(LONG_POLL_DIRECTIVE, &format!(" /// {directive}\n")) diff --git a/xtask/src/build_examples.rs b/xtask/src/build_examples.rs index 8623717e8..21cbc4f5a 100644 --- a/xtask/src/build_examples.rs +++ b/xtask/src/build_examples.rs @@ -23,6 +23,8 @@ struct ExampleConfig { languages: Option>, #[serde(default)] system_nexus: bool, + #[serde(default)] + client: bool, } impl ExampleConfig { @@ -442,14 +444,13 @@ fn build_example(repo_root: &Path, language: Language, example_id: &str) -> Resu let output_path = example_output_path(repo_root, language, example_id); let language_root = advanced_language_root(repo_root, language); reset_example_output_directory(&language_root, &output_path)?; - let system_nexus = config - .examples - .get(example_id) - .is_some_and(|example| example.system_nexus); + let example = config.examples.get(example_id); + let system_nexus = example.is_some_and(|example| example.system_nexus); let generate_request = GenerateRequest { config: NexgenConfig { mode: GenerationMode::NativeApi, system_nexus, + client: example.is_some_and(|example| example.client), }, language, input_paths,