From 93e862fac28888697ca80d8a4f319e213c214e6d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 19:40:40 -0700 Subject: [PATCH 1/6] Added the @nexus.one-way operation marker with a sync Python stub. --- GUIDE.md | 23 ++++ changelog/added/paper-plane-penguin.md | 1 + src/error.rs | 8 ++ src/generator/python.rs | 145 ++++++++++++++++++++++++- src/parser/json_schema.rs | 1 + src/parser/wit.rs | 38 +++++++ src/planning/type_planning.rs | 2 + src/spec.rs | 5 + tests/wit_streaming.rs | 73 ++++++++++++- 9 files changed, 293 insertions(+), 3 deletions(-) create mode 100644 changelog/added/paper-plane-penguin.md diff --git a/GUIDE.md b/GUIDE.md index 9d42854f5..3b50821f0 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -1798,6 +1798,29 @@ variant notification-target { --- +### @nexus.one-way + +**Placement:** Operation (function) **Syntax:** `@nexus.one-way` + +Marks an operation whose result the Workflow caller does not use, such as an +append that a plugin buffers. The directive takes no arguments. + +**Python:** the Workflow stubs are synchronous and return `None`. Each stub +still starts the operation through the Workflow's Nexus client, so every +outbound interceptor sees the call. An interceptor must take the call and +complete it without suspending, for example by buffering it and returning a done +handle. If the start suspends, the stub raises `RuntimeError`. + +```python +def append_records(*, stream: StreamRef | None = None, ...) -> None: ... +``` + +TypeScript and .NET accept the directive and keep their asynchronous stubs. +Cannot be combined with `@nexus.long-poll` or `@nexus.output-transform`, and the +operation cannot return a resource. + +--- + ### @nexus.operation **Placement:** Operation (function) **Syntax:** diff --git a/changelog/added/paper-plane-penguin.md b/changelog/added/paper-plane-penguin.md new file mode 100644 index 000000000..d5e7fa12c --- /dev/null +++ b/changelog/added/paper-plane-penguin.md @@ -0,0 +1 @@ +WIT operations accept `@nexus.one-way`. Its Python Workflow stubs are synchronous and return `None`, and they still start the operation through the outbound interceptors. diff --git a/src/error.rs b/src/error.rs index 78995c800..46f7c24a0 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 one-way operation `{service}.{operation}`: {reason}")] + UnsupportedOneWayOperation { + 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/python.rs b/src/generator/python.rs index d6b229599..47561dab9 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -490,7 +490,7 @@ impl<'a> ApiPlanner<'a> { let operations = service .operations .iter() - .map(|operation| self.resolve_operation(operation)) + .map(|operation| self.resolve_operation(&service.name, operation)) .collect::>>()?; Ok(RenderedService { @@ -1139,6 +1139,7 @@ impl<'a> ApiPlanner<'a> { fn resolve_operation<'operation>( &mut self, + service_name: &str, operation: &'operation OperationSpec, ) -> Result> { let output_resource_return = operation.data.output_resource_return.clone(); @@ -1271,6 +1272,26 @@ impl<'a> ApiPlanner<'a> { &local_python_model_type_expr(&output_ref), &output_type_parameters, ); + if operation.one_way { + let reason = if generic_model_types { + Some("its models take type parameters") + } else if unpacked_input + .as_ref() + .is_some_and(|unpacked_input| !unpacked_input.functions.is_empty()) + { + Some("its input has function fields") + } else { + None + }; + if let Some(reason) = reason { + return Err(Error::UnsupportedOneWayOperation { + language: Language::Python, + service: service_name.to_string(), + operation: operation.name.clone(), + reason: reason.to_string(), + }); + } + } let descriptor_output_ref = match operation.output_type() { Some(PlannedType::Record(record)) => python_erased_generic_record_annotation( &output_ref, @@ -1313,6 +1334,7 @@ impl<'a> ApiPlanner<'a> { .map(|usage| usage.parameter.name) .collect(), generic_model_types, + one_way: operation.one_way, }) } @@ -2543,6 +2565,7 @@ struct RenderedOperation<'a> { unpacked_input: Option, model_input_type_parameters: Vec, generic_model_types: bool, + one_way: bool, } #[derive(Debug)] @@ -6587,6 +6610,10 @@ fn render_operation_functions( service: &RenderedService<'_>, operation: &RenderedOperation<'_>, ) { + if operation.one_way { + render_one_way_operation_functions(output, service, operation); + return; + } render_request_only_operation_function(output, service, operation); if service.endpoint.is_none() { return; @@ -6601,6 +6628,122 @@ fn render_operation_functions( } } +/// A one-way stub is synchronous, because Workflow code does not wait for a +/// result it does not use. It still starts the operation through the Nexus +/// client, so every outbound interceptor sees the call. The interceptor that +/// takes it, such as a plugin that buffers it, must complete it without +/// suspending. +fn render_one_way_operation_functions( + output: &mut String, + service: &RenderedService<'_>, + operation: &RenderedOperation<'_>, +) { + let function_name = request_operation_function_name(service, operation); + render_deprecated_decorator(output, operation); + output.push_str("def "); + output.push_str(&function_name); + output.push_str("(\n"); + render_endpoint_parameter(output, service); + if let Some(input) = &operation.input { + output.push_str(" request: "); + output.push_str(&input.annotation); + output.push_str(",\n"); + } + output.push_str(") -> None:\n"); + if operation.unpacked_input.is_none() || service.endpoint.is_none() { + render_operation_docstring(output, operation, &[], None); + } + render_temporalio_workflow_runtime_imports( + output, + service, + &temporalio_workflow_operation_runtime_names(service, operation), + " ", + ); + render_inline_nexus_client(output, service, " "); + output.push_str(" start = nexus_client.start_operation(\n"); + output.push_str(" operation="); + output.push_str(&python_string_literal(operation.wire_name)); + output.push_str(",\n"); + output.push_str(" input="); + output.push_str(operation_input_wire_expr(operation)); + output.push_str(",\n"); + if !operation.output_none { + output.push_str(" output_type="); + output.push_str(&operation.output_type_expr); + output.push_str(",\n"); + } + output.push_str(" )\n"); + output.push_str(" try:\n"); + output.push_str(" _ = start.send(None)\n"); + output.push_str(" except StopIteration:\n"); + output.push_str(" return\n"); + output.push_str(" start.close()\n"); + output.push_str(" raise RuntimeError(\n"); + output.push_str(" "); + output.push_str(&python_string_literal(&format!( + "one-way operation {} did not complete at once. ", + operation.wire_name + ))); + output.push_str( + "\n \"It needs an outbound interceptor that takes the call, such as one that \"\n", + ); + output.push_str(" \"buffers it.\"\n"); + output.push_str(" )\n"); + + if service.endpoint.is_none() { + return; + } + let Some(unpacked_input) = &operation.unpacked_input else { + return; + }; + output.push('\n'); + render_deprecated_decorator(output, operation); + output.push_str("def "); + output.push_str(&operation.attr_name); + output.push_str("(\n"); + output.push_str(" *,\n"); + for field in &unpacked_input.parameters { + output.push_str(" "); + output.push_str(&field.attr_name); + output.push_str(": "); + output.push_str(&python_parameter_annotation( + &field.annotation, + &field.default_kind, + )); + if let Some(default_expr) = python_parameter_default_expr(&field.default_kind) { + render_python_default_expr(output, &default_expr, " "); + } + output.push_str(",\n"); + } + output.push_str(") -> None:\n"); + render_unpacked_operation_docstring(output, operation, unpacked_input, None); + for flattened_message in &unpacked_input.flattened_messages { + render_flattened_message_setup(output, flattened_message); + } + output.push_str(" request = "); + output.push_str(&unpacked_input.model_name); + output.push_str("(\n"); + for field in &unpacked_input.request_fields { + output.push_str(" "); + output.push_str(&field.attr_name); + output.push('='); + output.push_str(&unpacked_request_field_value_expr(field)); + output.push_str(",\n"); + } + output.push_str(" )\n"); + output.push_str(" "); + output.push_str(&function_name); + output.push_str("(request)\n"); +} + +fn render_deprecated_decorator(output: &mut String, operation: &RenderedOperation<'_>) { + if operation.deprecated { + output.push_str( + "@typing_extensions.deprecated(\"This operation is deprecated.\", category=None)\n", + ); + } +} + fn function_unpacked_needs_overloads(functions: &[RenderedFunctionField]) -> bool { functions.iter().any(|function| { function.alternate_annotation.is_some() diff --git a/src/parser/json_schema.rs b/src/parser/json_schema.rs index 6bb0821f0..6377f6cdb 100644 --- a/src/parser/json_schema.rs +++ b/src/parser/json_schema.rs @@ -6177,6 +6177,7 @@ fn build_operation( output_transform: None, serialization_context: LanguageStringSpec::default(), long_poll: None, + one_way: false, data: (), }) } diff --git a/src/parser/wit.rs b/src/parser/wit.rs index 25e5599e9..56befd206 100644 --- a/src/parser/wit.rs +++ b/src/parser/wit.rs @@ -2965,6 +2965,28 @@ fn build_operation( path, &context, )?; + let one_way = one_way_directive(&directives, path, &context)?; + if one_way { + let reject = |reason: &str| Error::InvalidWitDirective { + path: path.to_path_buf(), + context: context.clone(), + directive: "@nexus.one-way".to_string(), + reason: reason.to_string(), + }; + if output_transform.is_some() { + return Err(reject("cannot be combined with `@nexus.output-transform`")); + } + if long_poll.is_some() { + return Err(reject("cannot be combined with `@nexus.long-poll`")); + } + if function.result.as_ref().is_some_and(|output_type| { + find_owned_resource_name_for_type(resolve, output_type).is_some() + }) { + return Err(reject( + "cannot return a resource, since the caller drops the result", + )); + } + } let output = if let Some(output_type) = function.result.as_ref() { Some( find_operation_type_spec(resolve, output_type, path, &context)?.ok_or_else(|| { @@ -3005,6 +3027,7 @@ fn build_operation( output_transform, serialization_context, long_poll, + one_way, data: (), }) } @@ -3663,6 +3686,21 @@ fn is_integer_type(resolve: &Resolve, ty: &Type) -> bool { } } +fn one_way_directive(directives: &[Directive], path: &Path, context: &str) -> Result { + let Some(directive) = directive(directives, "one-way", path, context)? else { + return Ok(false); + }; + if !directive.args.is_empty() { + return Err(Error::InvalidWitDirective { + path: path.to_path_buf(), + context: context.to_string(), + directive: "@nexus.one-way".to_string(), + reason: "does not take arguments".to_string(), + }); + } + Ok(true) +} + fn delay_load_temporalio_workflow_directive( directives: &[Directive], path: &Path, diff --git a/src/planning/type_planning.rs b/src/planning/type_planning.rs index c9345a172..96a59fab7 100644 --- a/src/planning/type_planning.rs +++ b/src/planning/type_planning.rs @@ -656,6 +656,7 @@ impl<'a> TypePlanningContext<'a> { }), serialization_context: operation.serialization_context.clone(), long_poll: operation.long_poll.clone(), + one_way: operation.one_way, data: PlannedOperationData { output_resource_return: plan_operation_resource_return( operation.data.output_resource_return.as_ref(), @@ -1721,6 +1722,7 @@ mod tests { output_transform: None, serialization_context: LanguageStringSpec::default(), long_poll: None, + one_way: false, data: (), }], resources: Vec::new(), diff --git a/src/spec.rs b/src/spec.rs index b627f0dad..cc715a8d0 100644 --- a/src/spec.rs +++ b/src/spec.rs @@ -705,6 +705,10 @@ pub struct OperationSpec { pub serialization_context: F::Text, /// The long-poll contract, when the operation declares one. pub long_poll: Option, + /// Whether the caller in a Workflow does not use the result + /// (`@nexus.one-way` on WIT). The Workflow stub is then synchronous and + /// returns nothing. The call still goes through the outbound interceptors. + pub one_way: bool, pub data: F::OperationData, } @@ -742,6 +746,7 @@ impl OperationSpec { .map(|transform| transform.map_names_with(map)), serialization_context: map.map_text(self.serialization_context), long_poll: self.long_poll, + one_way: self.one_way, data, } } diff --git a/tests/wit_streaming.rs b/tests/wit_streaming.rs index 1ed3412ec..35ac8f716 100644 --- a/tests/wit_streaming.rs +++ b/tests/wit_streaming.rs @@ -1,5 +1,5 @@ -//! 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` and `@nexus.one-way` on operations. #![cfg(feature = "advanced")] use std::collections::BTreeMap; @@ -60,6 +60,7 @@ interface log-service { done: option, } + /// @nexus.one-way append: func(request: append-request) -> append-response; /// @nexus.long-poll @@ -382,3 +383,71 @@ fn long_poll_rejects_an_operation_whose_output_is_not_a_record() { "{error}" ); } + +#[test] +fn python_one_way_stub_is_synchronous_and_returns_nothing() { + let files = generate(Language::Python, GenerationMode::NativeApi, LOG_WIT); + let append = file(&files, "operations/append.py"); + + assert!( + append.contains("\ndef _append(\n request: AppendRequest,\n) -> None:\n"), + "{append}" + ); + assert!(append.contains("\ndef append(\n *,\n"), "{append}"); + assert!(!append.contains("async def"), "{append}"); + assert!(!append.contains("NexusOperationHandle"), "{append}"); + // The call goes through the Nexus client, so the outbound interceptors see + // it, and the stub drives the start to its end without suspending. + assert!( + append.contains(" start = nexus_client.start_operation(\n"), + "{append}" + ); + assert!( + append + .contains(" _ = start.send(None)\n except StopIteration:\n return\n"), + "{append}" + ); + assert!( + append.contains(" start.close()\n raise RuntimeError("), + "{append}" + ); + assert!(append.contains(" _append(request)\n"), "{append}"); + + let read = file(&files, "operations/read.py"); + assert!(read.contains("async def read("), "{read}"); +} + +#[test] +fn one_way_generates_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 one_way_rejects_arguments_and_conflicting_directives() { + for (from, to, reason) in [ + ( + " /// @nexus.one-way\n", + " /// @nexus.one-way buffered=true\n", + "does not take arguments", + ), + ( + " /// @nexus.long-poll\n", + " /// @nexus.one-way\n /// @nexus.long-poll\n", + "cannot be combined with `@nexus.long-poll`", + ), + ( + " /// @nexus.one-way\n", + " /// @nexus.one-way\n /// @nexus.output-transform python-type=\"int\" python=\"1\"\n", + "cannot be combined with `@nexus.output-transform`", + ), + ] { + let error = generate_error(Language::Python, &LOG_WIT.replacen(from, to, 1)); + assert!( + error.contains("@nexus.one-way") && error.contains(reason), + "{to}: {error}" + ); + } +} From 3539efae18514d651f24f484e829f1dd97164d1e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 19:48:08 -0700 Subject: [PATCH 2/6] Generated a Python caller per service over a replaceable transport. --- GUIDE.md | 39 ++++ advanced/README.md | 7 +- changelog/added/relay-raccoon.md | 1 + src/error.rs | 8 + src/generator/go.rs | 1 + src/generator/python.rs | 328 +++++++++++++++++++++++++++++++ src/main.rs | 9 + src/nexgen_config.rs | 6 + tests/generate_go.rs | 2 + tests/wit_streaming.rs | 163 ++++++++++++++- xtask/src/build_examples.rs | 1 + 11 files changed, 553 insertions(+), 12 deletions(-) create mode 100644 changelog/added/relay-raccoon.md diff --git a/GUIDE.md b/GUIDE.md index 3b50821f0..57cf3f3c4 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) @@ -1044,6 +1045,44 @@ 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`. Other models go through the payload converter that the caller +takes (the SDK default when none is given). A `payload_codec`, when given, +encodes each request payload and decodes each answer. A transport raises +`TransportError` for a call that failed, with `retryable` set when the same call +can succeed later. + +```python +client = StreamServiceClient(my_transport) +response = await client.describe_stream(DescribeStreamRequest(stream=ref)) +``` + +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 757888cd8..4e383b5c0 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/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 46f7c24a0..a83d4cae7 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("{language} cannot generate one-way operation `{service}.{operation}`: {reason}")] UnsupportedOneWayOperation { language: Language, diff --git a/src/generator/go.rs b/src/generator/go.rs index f249bebbe..71a16b141 100644 --- a/src/generator/go.rs +++ b/src/generator/go.rs @@ -5882,6 +5882,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 47561dab9..34ca2628b 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", @@ -1335,9 +1342,23 @@ impl<'a> ApiPlanner<'a> { .collect(), generic_model_types, one_way: operation.one_way, + 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(_)) @@ -2566,6 +2587,11 @@ struct RenderedOperation<'a> { model_input_type_parameters: Vec, generic_model_types: bool, one_way: 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)] @@ -4009,6 +4035,308 @@ fn render_models_module( Ok(Some(output)) } +/// The model a caller outside a Workflow names 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( + " payload_converter: temporalio.converter.PayloadConverter | None = None,\n", + ); + body.push_str(" payload_codec: temporalio.converter.PayloadCodec | 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( + " payload_converter: Converts the models that are not proto-backed.\n", + ); + body.push_str(" Defaults to the SDK's default payload converter.\n"); + body.push_str( + " payload_codec: Encodes each request payload and decodes each answer.\n", + ); + body.push_str(" \"\"\"\n"); + body.push_str(" self._transport = transport\n"); + body.push_str(" self._payload_converter = (\n"); + body.push_str(" payload_converter or temporalio.converter.DataConverter.default.payload_converter\n"); + body.push_str(" )\n"); + body.push_str(" self._payload_codec = payload_codec\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(" if self._payload_codec is not None:\n"); + body.push_str(" [payload] = await self._payload_codec.encode([payload])\n"); + body.push_str(" answer = 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"); + body.push_str(" if self._payload_codec is not None:\n"); + body.push_str(" [answer] = await self._payload_codec.decode([answer])\n"); + body.push_str(" return answer\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()); + } + } + + let mut output = String::new(); + render_generated_file_header(&mut output); + output.push('\n'); + output.push_str("import typing\n\n"); + 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('\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, + ); + output.push_str(" payload = "); + match &operation.client_input_proto_model { + Some(proto_model) => { + output.push_str("_proto_payload(_"); + output.push_str(proto_model); + output.push_str("TransferTypeConverter().to_transfer_type(request))\n"); + } + None => output.push_str("self._payload_converter.to_payload(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(" return converter.from_transfer_type(\n"); + output.push_str(" _proto_message(answer, converter.transfer_type),\n"); + output.push_str(" "); + output.push_str(output_model); + output.push_str(",\n"); + output.push_str(" )\n"); + } + None => { + output.push_str(" return typing.cast(\n"); + output.push_str(" "); + output.push_str(output_model); + output.push_str(",\n"); + output.push_str(" self._payload_converter.from_payload(answer, "); + output.push_str(output_model); + output.push_str("),\n"); + output.push_str(" )\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 + + +_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 +"#; + 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 07fa4d424..818f18bc4 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(), @@ -1497,6 +1498,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 35ac8f716..d5a057d17 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` and `@nexus.one-way` on operations. +//! WIT annotations for streaming services (`@nexus.offset` on record fields, +//! and `@nexus.long-poll` and `@nexus.one-way` on operations) and the caller +//! that `--client` generates for code outside a Workflow. #![cfg(feature = "advanced")] use std::collections::BTreeMap; @@ -123,22 +124,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", @@ -451,3 +462,135 @@ fn one_way_rejects_arguments_and_conflicting_directives() { ); } } + +fn client_config() -> NexgenConfig { + NexgenConfig { + mode: GenerationMode::NativeApi, + client: true, + ..Default::default() + } +} + +fn generate_client(wit: &str) -> String { + let files = try_generate_with(client_config(), Language::Python, wit, Vec::new()).unwrap(); + file(&files, "client.py").to_string() +} + +#[test] +fn python_client_calls_each_operation_over_a_transport() { + let client = generate_client(LOG_WIT); + + assert!( + client.contains("class Transport(typing.Protocol):"), + "{client}" + ); + assert!( + client.contains( + " async def call(\n self,\n *,\n service: str,\n \ + operation: str,\n input: temporalio.api.common.v1.Payload,\n ) -> \ + temporalio.api.common.v1.Payload:" + ), + "{client}" + ); + assert!( + client.contains("class TransportError(Exception):"), + "{client}" + ); + assert!(client.contains("class LogServiceClient:"), "{client}"); + assert!( + client.contains(" service=\"LogService\",\n operation=operation,"), + "{client}" + ); + // A model that is not proto-backed goes through the payload converter. + assert!( + client.contains( + " async def append(self, request: AppendRequest) -> AppendResponse:\n \ + payload = self._payload_converter.to_payload(request)\n \ + answer = await self._call(\"Append\", payload)\n" + ), + "{client}" + ); + // A one-way operation is synchronous in a Workflow only. Outside one, the + // caller still answers with the result. + assert!( + client.contains(" async def read(self, request: ReadRequest) -> ReadResponse:"), + "{client}" + ); + assert!( + client.contains("[payload] = await self._payload_codec.encode([payload])"), + "{client}" + ); + assert!( + client.contains("[answer] = await self._payload_codec.decode([answer])"), + "{client}" + ); +} + +#[test] +fn python_client_converts_proto_backed_models_through_their_converters() { + let root = project_root(); + let files = try_generate_with( + client_config(), + Language::Python, + &fs::read_to_string(root.join("advanced/samples/inputs/start-workflow.wit")).unwrap(), + Vec::new(), + ) + .unwrap(); + let client = file(&files, "client.py"); + + assert!( + client.contains( + " payload = _proto_payload(_CancelWorkflowRequestTransferTypeConverter().to_transfer_type(request))\n" + ), + "{client}" + ); + assert!( + client.contains( + " converter = _CancelWorkflowResponseTransferTypeConverter()\n \ + return converter.from_transfer_type(\n \ + _proto_message(answer, converter.transfer_type),\n CancelWorkflowResponse,\n )\n" + ), + "{client}" + ); + assert!( + client.contains(" _CancelWorkflowRequestTransferTypeConverter,\n"), + "{client}" + ); +} + +#[test] +fn python_client_is_generated_only_with_the_client_flag() { + let files = generate(Language::Python, GenerationMode::NativeApi, LOG_WIT); + assert!(!files.contains_key("client.py"), "{:?}", files.keys()); +} + +#[test] +fn python_client_rejects_an_operation_that_returns_a_resource_without_a_wire_type() { + let root = project_root(); + let error = try_generate_with( + client_config(), + Language::Python, + &fs::read_to_string(root.join("advanced/samples/inputs/user-service.wit")).unwrap(), + Vec::new(), + ) + .unwrap_err(); + assert!( + error.contains("cannot generate a caller for operation `UserService.GetUser`") + && error.contains("it returns a resource"), + "{error}" + ); +} + +#[test] +fn client_flag_generates_nothing_more_in_typescript_and_dotnet() { + for language in [Language::TypeScript, Language::Dotnet] { + let with_client = + try_generate_with(client_config(), language, LOG_WIT, Vec::new()).unwrap(); + let without_client = generate(language, GenerationMode::NativeApi, LOG_WIT); + assert_eq!( + with_client.keys().collect::>(), + without_client.keys().collect::>(), + "{language:?}" + ); + } +} diff --git a/xtask/src/build_examples.rs b/xtask/src/build_examples.rs index d78b0fade..a3d35600c 100644 --- a/xtask/src/build_examples.rs +++ b/xtask/src/build_examples.rs @@ -450,6 +450,7 @@ fn build_example(repo_root: &Path, language: Language, example_id: &str) -> Resu config: NexgenConfig { mode: GenerationMode::NativeApi, system_nexus, + client: false, }, language, input_paths, From 19a92c8e05813c444cbc8d14be96df9ae1f148c8 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 21 Oct 2026 23:48:26 -0700 Subject: [PATCH 3/6] Reverted the @nexus.one-way operation marker. The Stream WIT no longer marks any operation one-way, so the parser, generator, tests, docs and changelog fragment go. --- GUIDE.md | 23 ---- changelog/added/paper-plane-penguin.md | 1 - src/error.rs | 8 -- src/generator/python.rs | 145 +------------------------ src/parser/json_schema.rs | 1 - src/parser/wit.rs | 38 ------- src/planning/type_planning.rs | 2 - src/spec.rs | 5 - tests/wit_streaming.rs | 73 +------------ 9 files changed, 3 insertions(+), 293 deletions(-) delete mode 100644 changelog/added/paper-plane-penguin.md diff --git a/GUIDE.md b/GUIDE.md index 3b50821f0..9d42854f5 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -1798,29 +1798,6 @@ variant notification-target { --- -### @nexus.one-way - -**Placement:** Operation (function) **Syntax:** `@nexus.one-way` - -Marks an operation whose result the Workflow caller does not use, such as an -append that a plugin buffers. The directive takes no arguments. - -**Python:** the Workflow stubs are synchronous and return `None`. Each stub -still starts the operation through the Workflow's Nexus client, so every -outbound interceptor sees the call. An interceptor must take the call and -complete it without suspending, for example by buffering it and returning a done -handle. If the start suspends, the stub raises `RuntimeError`. - -```python -def append_records(*, stream: StreamRef | None = None, ...) -> None: ... -``` - -TypeScript and .NET accept the directive and keep their asynchronous stubs. -Cannot be combined with `@nexus.long-poll` or `@nexus.output-transform`, and the -operation cannot return a resource. - ---- - ### @nexus.operation **Placement:** Operation (function) **Syntax:** diff --git a/changelog/added/paper-plane-penguin.md b/changelog/added/paper-plane-penguin.md deleted file mode 100644 index d5e7fa12c..000000000 --- a/changelog/added/paper-plane-penguin.md +++ /dev/null @@ -1 +0,0 @@ -WIT operations accept `@nexus.one-way`. Its Python Workflow stubs are synchronous and return `None`, and they still start the operation through the outbound interceptors. diff --git a/src/error.rs b/src/error.rs index 46f7c24a0..78995c800 100644 --- a/src/error.rs +++ b/src/error.rs @@ -215,14 +215,6 @@ pub enum Error { )] IncompleteOperationOutputTransform { service: String, operation: String }, - #[error("{language} cannot generate one-way operation `{service}.{operation}`: {reason}")] - UnsupportedOneWayOperation { - 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/python.rs b/src/generator/python.rs index 74270e3e9..930fc19d6 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -490,7 +490,7 @@ impl<'a> ApiPlanner<'a> { let operations = service .operations .iter() - .map(|operation| self.resolve_operation(&service.name, operation)) + .map(|operation| self.resolve_operation(operation)) .collect::>>()?; Ok(RenderedService { @@ -1139,7 +1139,6 @@ impl<'a> ApiPlanner<'a> { fn resolve_operation<'operation>( &mut self, - service_name: &str, operation: &'operation OperationSpec, ) -> Result> { let output_resource_return = operation.data.output_resource_return.clone(); @@ -1272,26 +1271,6 @@ impl<'a> ApiPlanner<'a> { &local_python_model_type_expr(&output_ref), &output_type_parameters, ); - if operation.one_way { - let reason = if generic_model_types { - Some("its models take type parameters") - } else if unpacked_input - .as_ref() - .is_some_and(|unpacked_input| !unpacked_input.functions.is_empty()) - { - Some("its input has function fields") - } else { - None - }; - if let Some(reason) = reason { - return Err(Error::UnsupportedOneWayOperation { - language: Language::Python, - service: service_name.to_string(), - operation: operation.name.clone(), - reason: reason.to_string(), - }); - } - } let descriptor_output_ref = match operation.output_type() { Some(PlannedType::Record(record)) => python_erased_generic_record_annotation( &output_ref, @@ -1334,7 +1313,6 @@ impl<'a> ApiPlanner<'a> { .map(|usage| usage.parameter.name) .collect(), generic_model_types, - one_way: operation.one_way, }) } @@ -2568,7 +2546,6 @@ struct RenderedOperation<'a> { unpacked_input: Option, model_input_type_parameters: Vec, generic_model_types: bool, - one_way: bool, } #[derive(Debug)] @@ -6627,10 +6604,6 @@ fn render_operation_functions( service: &RenderedService<'_>, operation: &RenderedOperation<'_>, ) { - if operation.one_way { - render_one_way_operation_functions(output, service, operation); - return; - } render_request_only_operation_function(output, service, operation); if service.endpoint.is_none() { return; @@ -6645,122 +6618,6 @@ fn render_operation_functions( } } -/// A one-way stub is synchronous, because Workflow code does not wait for a -/// result it does not use. It still starts the operation through the Nexus -/// client, so every outbound interceptor sees the call. The interceptor that -/// takes it, such as a plugin that buffers it, must complete it without -/// suspending. -fn render_one_way_operation_functions( - output: &mut String, - service: &RenderedService<'_>, - operation: &RenderedOperation<'_>, -) { - let function_name = request_operation_function_name(service, operation); - render_deprecated_decorator(output, operation); - output.push_str("def "); - output.push_str(&function_name); - output.push_str("(\n"); - render_endpoint_parameter(output, service); - if let Some(input) = &operation.input { - output.push_str(" request: "); - output.push_str(&input.annotation); - output.push_str(",\n"); - } - output.push_str(") -> None:\n"); - if operation.unpacked_input.is_none() || service.endpoint.is_none() { - render_operation_docstring(output, operation, &[], None); - } - render_temporalio_workflow_runtime_imports( - output, - service, - &temporalio_workflow_operation_runtime_names(service, operation), - " ", - ); - render_inline_nexus_client(output, service, " "); - output.push_str(" start = nexus_client.start_operation(\n"); - output.push_str(" operation="); - output.push_str(&python_string_literal(operation.wire_name)); - output.push_str(",\n"); - output.push_str(" input="); - output.push_str(operation_input_wire_expr(operation)); - output.push_str(",\n"); - if !operation.output_none { - output.push_str(" output_type="); - output.push_str(&operation.output_type_expr); - output.push_str(",\n"); - } - output.push_str(" )\n"); - output.push_str(" try:\n"); - output.push_str(" _ = start.send(None)\n"); - output.push_str(" except StopIteration:\n"); - output.push_str(" return\n"); - output.push_str(" start.close()\n"); - output.push_str(" raise RuntimeError(\n"); - output.push_str(" "); - output.push_str(&python_string_literal(&format!( - "one-way operation {} did not complete at once. ", - operation.wire_name - ))); - output.push_str( - "\n \"It needs an outbound interceptor that takes the call, such as one that \"\n", - ); - output.push_str(" \"buffers it.\"\n"); - output.push_str(" )\n"); - - if service.endpoint.is_none() { - return; - } - let Some(unpacked_input) = &operation.unpacked_input else { - return; - }; - output.push('\n'); - render_deprecated_decorator(output, operation); - output.push_str("def "); - output.push_str(&operation.attr_name); - output.push_str("(\n"); - output.push_str(" *,\n"); - for field in &unpacked_input.parameters { - output.push_str(" "); - output.push_str(&field.attr_name); - output.push_str(": "); - output.push_str(&python_parameter_annotation( - &field.annotation, - &field.default_kind, - )); - if let Some(default_expr) = python_parameter_default_expr(&field.default_kind) { - render_python_default_expr(output, &default_expr, " "); - } - output.push_str(",\n"); - } - output.push_str(") -> None:\n"); - render_unpacked_operation_docstring(output, operation, unpacked_input, None); - for flattened_message in &unpacked_input.flattened_messages { - render_flattened_message_setup(output, flattened_message); - } - output.push_str(" request = "); - output.push_str(&unpacked_input.model_name); - output.push_str("(\n"); - for field in &unpacked_input.request_fields { - output.push_str(" "); - output.push_str(&field.attr_name); - output.push('='); - output.push_str(&unpacked_request_field_value_expr(field)); - output.push_str(",\n"); - } - output.push_str(" )\n"); - output.push_str(" "); - output.push_str(&function_name); - output.push_str("(request)\n"); -} - -fn render_deprecated_decorator(output: &mut String, operation: &RenderedOperation<'_>) { - if operation.deprecated { - output.push_str( - "@typing_extensions.deprecated(\"This operation is deprecated.\", category=None)\n", - ); - } -} - fn function_unpacked_needs_overloads(functions: &[RenderedFunctionField]) -> bool { functions.iter().any(|function| { function.alternate_annotation.is_some() diff --git a/src/parser/json_schema.rs b/src/parser/json_schema.rs index 6377f6cdb..6bb0821f0 100644 --- a/src/parser/json_schema.rs +++ b/src/parser/json_schema.rs @@ -6177,7 +6177,6 @@ fn build_operation( output_transform: None, serialization_context: LanguageStringSpec::default(), long_poll: None, - one_way: false, data: (), }) } diff --git a/src/parser/wit.rs b/src/parser/wit.rs index 56befd206..25e5599e9 100644 --- a/src/parser/wit.rs +++ b/src/parser/wit.rs @@ -2965,28 +2965,6 @@ fn build_operation( path, &context, )?; - let one_way = one_way_directive(&directives, path, &context)?; - if one_way { - let reject = |reason: &str| Error::InvalidWitDirective { - path: path.to_path_buf(), - context: context.clone(), - directive: "@nexus.one-way".to_string(), - reason: reason.to_string(), - }; - if output_transform.is_some() { - return Err(reject("cannot be combined with `@nexus.output-transform`")); - } - if long_poll.is_some() { - return Err(reject("cannot be combined with `@nexus.long-poll`")); - } - if function.result.as_ref().is_some_and(|output_type| { - find_owned_resource_name_for_type(resolve, output_type).is_some() - }) { - return Err(reject( - "cannot return a resource, since the caller drops the result", - )); - } - } let output = if let Some(output_type) = function.result.as_ref() { Some( find_operation_type_spec(resolve, output_type, path, &context)?.ok_or_else(|| { @@ -3027,7 +3005,6 @@ fn build_operation( output_transform, serialization_context, long_poll, - one_way, data: (), }) } @@ -3686,21 +3663,6 @@ fn is_integer_type(resolve: &Resolve, ty: &Type) -> bool { } } -fn one_way_directive(directives: &[Directive], path: &Path, context: &str) -> Result { - let Some(directive) = directive(directives, "one-way", path, context)? else { - return Ok(false); - }; - if !directive.args.is_empty() { - return Err(Error::InvalidWitDirective { - path: path.to_path_buf(), - context: context.to_string(), - directive: "@nexus.one-way".to_string(), - reason: "does not take arguments".to_string(), - }); - } - Ok(true) -} - fn delay_load_temporalio_workflow_directive( directives: &[Directive], path: &Path, diff --git a/src/planning/type_planning.rs b/src/planning/type_planning.rs index 96a59fab7..c9345a172 100644 --- a/src/planning/type_planning.rs +++ b/src/planning/type_planning.rs @@ -656,7 +656,6 @@ impl<'a> TypePlanningContext<'a> { }), serialization_context: operation.serialization_context.clone(), long_poll: operation.long_poll.clone(), - one_way: operation.one_way, data: PlannedOperationData { output_resource_return: plan_operation_resource_return( operation.data.output_resource_return.as_ref(), @@ -1722,7 +1721,6 @@ mod tests { output_transform: None, serialization_context: LanguageStringSpec::default(), long_poll: None, - one_way: false, data: (), }], resources: Vec::new(), diff --git a/src/spec.rs b/src/spec.rs index cc715a8d0..b627f0dad 100644 --- a/src/spec.rs +++ b/src/spec.rs @@ -705,10 +705,6 @@ pub struct OperationSpec { pub serialization_context: F::Text, /// The long-poll contract, when the operation declares one. pub long_poll: Option, - /// Whether the caller in a Workflow does not use the result - /// (`@nexus.one-way` on WIT). The Workflow stub is then synchronous and - /// returns nothing. The call still goes through the outbound interceptors. - pub one_way: bool, pub data: F::OperationData, } @@ -746,7 +742,6 @@ impl OperationSpec { .map(|transform| transform.map_names_with(map)), serialization_context: map.map_text(self.serialization_context), long_poll: self.long_poll, - one_way: self.one_way, data, } } diff --git a/tests/wit_streaming.rs b/tests/wit_streaming.rs index 35ac8f716..1ed3412ec 100644 --- a/tests/wit_streaming.rs +++ b/tests/wit_streaming.rs @@ -1,5 +1,5 @@ -//! WIT annotations for streaming services: `@nexus.offset` on record fields, -//! and `@nexus.long-poll` and `@nexus.one-way` on operations. +//! WIT annotations for streaming services: `@nexus.offset` on record fields +//! and `@nexus.long-poll` on operations. #![cfg(feature = "advanced")] use std::collections::BTreeMap; @@ -60,7 +60,6 @@ interface log-service { done: option, } - /// @nexus.one-way append: func(request: append-request) -> append-response; /// @nexus.long-poll @@ -383,71 +382,3 @@ fn long_poll_rejects_an_operation_whose_output_is_not_a_record() { "{error}" ); } - -#[test] -fn python_one_way_stub_is_synchronous_and_returns_nothing() { - let files = generate(Language::Python, GenerationMode::NativeApi, LOG_WIT); - let append = file(&files, "operations/append.py"); - - assert!( - append.contains("\ndef _append(\n request: AppendRequest,\n) -> None:\n"), - "{append}" - ); - assert!(append.contains("\ndef append(\n *,\n"), "{append}"); - assert!(!append.contains("async def"), "{append}"); - assert!(!append.contains("NexusOperationHandle"), "{append}"); - // The call goes through the Nexus client, so the outbound interceptors see - // it, and the stub drives the start to its end without suspending. - assert!( - append.contains(" start = nexus_client.start_operation(\n"), - "{append}" - ); - assert!( - append - .contains(" _ = start.send(None)\n except StopIteration:\n return\n"), - "{append}" - ); - assert!( - append.contains(" start.close()\n raise RuntimeError("), - "{append}" - ); - assert!(append.contains(" _append(request)\n"), "{append}"); - - let read = file(&files, "operations/read.py"); - assert!(read.contains("async def read("), "{read}"); -} - -#[test] -fn one_way_generates_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 one_way_rejects_arguments_and_conflicting_directives() { - for (from, to, reason) in [ - ( - " /// @nexus.one-way\n", - " /// @nexus.one-way buffered=true\n", - "does not take arguments", - ), - ( - " /// @nexus.long-poll\n", - " /// @nexus.one-way\n /// @nexus.long-poll\n", - "cannot be combined with `@nexus.long-poll`", - ), - ( - " /// @nexus.one-way\n", - " /// @nexus.one-way\n /// @nexus.output-transform python-type=\"int\" python=\"1\"\n", - "cannot be combined with `@nexus.output-transform`", - ), - ] { - let error = generate_error(Language::Python, &LOG_WIT.replacen(from, to, 1)); - assert!( - error.contains("@nexus.one-way") && error.contains(reason), - "{to}: {error}" - ); - } -} From 37f8c517541f6163a3b8a23fed83d1b1fd2f1c7f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 11:32:09 -0700 Subject: [PATCH 4/6] Generated a Python caller per service over a transport. With --client, client.py has one caller per service for code outside a Workflow. The caller converts user values with a data converter, and a Transport carries each call. --- GUIDE.md | 41 ++ advanced/README.md | 7 +- advanced/samples/inputs/examples.toml | 3 + .../python/tests/test_start_workflow.py | 35 ++ .../python/wit/start_workflow/client.py | 230 ++++++++++ changelog/added/relay-raccoon.md | 1 + src/error.rs | 8 + src/generator/go.rs | 1 + src/generator/python.rs | 393 ++++++++++++++++++ src/main.rs | 9 + src/nexgen_config.rs | 6 + tests/generate_go.rs | 2 + xtask/src/build_examples.rs | 9 +- 13 files changed, 739 insertions(+), 6 deletions(-) create mode 100644 advanced/samples/python/wit/start_workflow/client.py create mode 100644 changelog/added/relay-raccoon.md diff --git a/GUIDE.md b/GUIDE.md index 79c2697ed..83cde3159 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 b18ca8054..e6063ce15 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -687,6 +687,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", @@ -1300,9 +1307,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(_)) @@ -2530,6 +2551,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)] @@ -3926,6 +3952,373 @@ fn render_models_module( Ok(Some(output)) } +/// The model a caller outside a Workflow names 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()); + } + } + + let mut output = String::new(); + render_generated_file_header(&mut output); + output.push('\n'); + output.push_str("import collections.abc\n"); + output.push_str("import contextlib\n"); + output.push_str("import typing\n\n"); + 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 + + +_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/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, From 225d2747b30c32c10cd994f5c91c498b601fab69 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 11:40:43 -0700 Subject: [PATCH 5/6] Emitted the proto helpers of a Python caller only when it uses them. --- src/generator/python.rs | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/generator/python.rs b/src/generator/python.rs index e6063ce15..62ab33fd8 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -4108,14 +4108,30 @@ fn render_client_module(services: &[RenderedService<'_>]) -> Result { } } + // 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'); - output.push_str("import collections.abc\n"); + if uses_proto { + output.push_str("import collections.abc\n"); + } output.push_str("import contextlib\n"); output.push_str("import typing\n\n"); - output.push_str("import google.protobuf.json_format\n"); - output.push_str("import google.protobuf.message\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"); @@ -4239,7 +4255,10 @@ class TransportError(Exception): """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) From b339830e41fd6a357d0fbce4e5457a0d8b7cbede Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 22 Oct 2026 13:05:37 -0700 Subject: [PATCH 6/6] Reworded the client model doc comment. --- src/generator/python.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/generator/python.rs b/src/generator/python.rs index 62ab33fd8..a5417dc59 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -3952,7 +3952,7 @@ fn render_models_module( Ok(Some(output)) } -/// The model a caller outside a Workflow names for an operation's input or +/// 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<'_>,