From 803aadb0751841ab93b2bb7836c14c55344da525 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 28 Oct 2026 12:02:57 -0700 Subject: [PATCH] feat(python): return a client call's result, with its RPC options Each client call and resource method takes `rpc_metadata` and `rpc_timeout`, and Workflow code refuses them. An `async func` operation also gets `start_`, which returns the handle. --- GUIDE.md | 17 +- advanced/samples/README.md | 10 + advanced/samples/inputs/client-calls.wit | 31 ++ advanced/samples/inputs/examples.toml | 6 + .../start-name-clash/expected.toml | 6 + .../client-target/start-name-clash/input.wit | 15 + .../samples/python/tests/test_client_calls.py | 103 ++++++ .../python/tests/test_client_target.py | 39 ++ .../samples/python/tests/test_poll_service.py | 5 +- .../python/wit/client_calls/__init__.py | 57 +++ .../samples/python/wit/client_calls/client.py | 181 ++++++++++ .../samples/python/wit/client_calls/models.py | 25 ++ .../wit/client_calls/operations/__init__.py | 3 + .../client_calls/operations/describe_job.py | 35 ++ .../wit/client_calls/operations/run_job.py | 35 ++ .../python/wit/client_calls/services.py | 24 ++ .../wit/multi_operation_service/client.py | 37 +- .../wit/poll_service/_resources/feed.py | 23 +- .../samples/python/wit/poll_service/client.py | 52 ++- .../wit/stream_service/_resources/stream.py | 68 +++- .../python/wit/stream_service/client.py | 66 +++- .../wit/stream_writer/_resources/writer.py | 47 ++- .../python/wit/stream_writer/client.py | 61 +++- .../python/wit/workflow_handle/client.py | 16 +- changelog/added/timely-tern.md | 2 + src/generator/python.rs | 340 ++++++++++++++++-- src/parser/json_schema.rs | 1 + src/parser/wit.rs | 1 + src/planning/type_planning.rs | 2 + src/spec.rs | 3 + 30 files changed, 1244 insertions(+), 67 deletions(-) create mode 100644 advanced/samples/inputs/client-calls.wit create mode 100644 advanced/samples/invalid/client-target/start-name-clash/expected.toml create mode 100644 advanced/samples/invalid/client-target/start-name-clash/input.wit create mode 100644 advanced/samples/python/tests/test_client_calls.py create mode 100644 advanced/samples/python/wit/client_calls/__init__.py create mode 100644 advanced/samples/python/wit/client_calls/client.py create mode 100644 advanced/samples/python/wit/client_calls/models.py create mode 100644 advanced/samples/python/wit/client_calls/operations/__init__.py create mode 100644 advanced/samples/python/wit/client_calls/operations/describe_job.py create mode 100644 advanced/samples/python/wit/client_calls/operations/run_job.py create mode 100644 advanced/samples/python/wit/client_calls/services.py create mode 100644 changelog/added/timely-tern.md diff --git a/GUIDE.md b/GUIDE.md index f8fbdf431..d02ac1de6 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -1490,13 +1490,20 @@ method returns an object with the `start_operation` of the Workflow Nexus client. A service with no `@nexus.endpoint` gets `endpoint=None`, and the client serves it. A call returns the result of the operation. +Each call takes the keyword options `rpc_metadata` and `rpc_timeout`, which +default to `None`, and passes them to `_create_nexus_client`. An operation that +WIT declares `async func` also has `start_`, which returns the handle of +the operation. The caller awaits the handle for the result later. The handle +gives the result of the operation as it is, so an operation that returns a +resource or has an output transform gets no `start_`. + ```python class StreamClient(StreamServiceClientExtension, StreamWriterClientExtension): - def _create_nexus_client(self, *, service, endpoint): + def _create_nexus_client(self, *, service, endpoint, rpc_metadata, rpc_timeout): ... # for example, start a standalone Nexus operation -stream = await StreamClient().create_stream(stream_id="s1") -await stream.publish(records=records) +stream = await StreamClient().create_stream(Turns, stream_id="s1") +await stream.publish(records=records, rpc_timeout=timedelta(seconds=10)) ``` Each operation also has a private method that takes the request, such as @@ -1515,6 +1522,10 @@ operations, such as one that you build from its fields, calls helper returns `workflow()`, the Workflow operations of a service with an endpoint. Outside a Workflow, it returns the current client. +The methods of the resource take the RPC options too. The Workflow operations +raise `ValueError` when either option is set, since the options apply to calls +outside a Workflow only. + The client target has no sources yet, so a `@nexus.source` field is a parameter of the client call. diff --git a/advanced/samples/README.md b/advanced/samples/README.md index ad6c813e5..cd8410c2e 100644 --- a/advanced/samples/README.md +++ b/advanced/samples/README.md @@ -81,6 +81,16 @@ parameter of an enum type. - Python: [`python/wit/poll_service/`](python/wit/poll_service/), [`python/tests/test_poll_service.py`](python/tests/test_poll_service.py) +### `client-calls` + +A service with an `async func` operation and a plain one. The Python client +target gives the `async` operation `start_run_job`, which returns the handle, +and each call takes the RPC options. + +- WIT: [`inputs/client-calls.wit`](inputs/client-calls.wit) +- Python: [`python/wit/client_calls/`](python/wit/client_calls/), + [`python/tests/test_client_calls.py`](python/tests/test_client_calls.py) + ### `workflow-handle` A System Nexus operation that returns a resource. Python generates only the diff --git a/advanced/samples/inputs/client-calls.wit b/advanced/samples/inputs/client-calls.wit new file mode 100644 index 000000000..55a8d7e08 --- /dev/null +++ b/advanced/samples/inputs/client-calls.wit @@ -0,0 +1,31 @@ +package temporal:client-calls@1.0.0; + +world service { + export job-service; +} + +// A client runs a short call to its end, or starts a long job and waits for it later. +/// @nexus.endpoint "jobs" +interface job-service { + record run-job-request { + name: string, + } + + record run-job-response { + output: string, + } + + record describe-job-request { + name: string, + } + + record describe-job-response { + status: string, + } + + /// @nexus.doc "Run a job, which can take a long time." + run-job: async func(request: run-job-request) -> run-job-response; + + /// @nexus.doc "Describe a job." + describe-job: func(request: describe-job-request) -> describe-job-response; +} diff --git a/advanced/samples/inputs/examples.toml b/advanced/samples/inputs/examples.toml index a3dd19642..63d87652e 100644 --- a/advanced/samples/inputs/examples.toml +++ b/advanced/samples/inputs/examples.toml @@ -66,6 +66,12 @@ languages = ["python"] targets = ["workflow", "client"] descriptors = "temporal" +# A client starts the `async` operation and waits for its result later. Only Python has the client +# target. +[client-calls] +languages = ["python"] +targets = ["workflow", "client"] + # A System Nexus operation that returns a resource. Python gets only the client target. [workflow-handle] languages = ["python", "dotnet", "typescript"] diff --git a/advanced/samples/invalid/client-target/start-name-clash/expected.toml b/advanced/samples/invalid/client-target/start-name-clash/expected.toml new file mode 100644 index 000000000..7af81c42f --- /dev/null +++ b/advanced/samples/invalid/client-target/start-name-clash/expected.toml @@ -0,0 +1,6 @@ +languages = ["python"] +targets = ["client"] + +[error] +kind = "client-target-name-conflict" +contains = ["`start_run_job` is a client method of both"] diff --git a/advanced/samples/invalid/client-target/start-name-clash/input.wit b/advanced/samples/invalid/client-target/start-name-clash/input.wit new file mode 100644 index 000000000..9c2d54f49 --- /dev/null +++ b/advanced/samples/invalid/client-target/start-name-clash/input.wit @@ -0,0 +1,15 @@ +package temporal:start-name-clash@1.0.0; + +world service { export job-service; } + +// The client starts `run-job` with `start_run_job`, which is also the name of the second operation. +/// @nexus.endpoint "jobs" +interface job-service { + record job-request { + name: string, + } + + run-job: async func(request: job-request); + + start-run-job: func(request: job-request); +} diff --git a/advanced/samples/python/tests/test_client_calls.py b/advanced/samples/python/tests/test_client_calls.py new file mode 100644 index 000000000..6b853b847 --- /dev/null +++ b/advanced/samples/python/tests/test_client_calls.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import collections.abc +import datetime +import typing + +from typing_extensions import override + +from wit.client_calls.client import JobServiceClientExtension +from wit.client_calls.models import ( + DescribeJobRequest, + DescribeJobResponse, + RunJobRequest, + RunJobResponse, +) + + +class FakeNexusClient: + def __init__(self, client: FakeClient) -> None: + self._client: FakeClient = client + + async def start_operation( + self, + operation: str, + input: typing.Any, + *, + output_type: type[typing.Any] | None = None, + ) -> collections.abc.Awaitable[typing.Any]: + del output_type + self._client.started.append((operation, input)) + answer = self._client.answers[operation] + + async def result() -> object: + self._client.awaited.append(operation) + return answer + + return result() + + +class FakeClient(JobServiceClientExtension): + def __init__(self, answers: dict[str, object]) -> None: + self.answers: dict[str, object] = answers + self.started: list[tuple[str, object]] = [] + self.awaited: list[str] = [] + self.rpc_options: list[ + tuple[ + collections.abc.Mapping[str, str | bytes] | None, + datetime.timedelta | None, + ] + ] = [] + + @override + def _create_nexus_client( + self, + *, + service: str, + endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, + ) -> FakeNexusClient: + del service, endpoint + self.rpc_options.append((rpc_metadata, rpc_timeout)) + return FakeNexusClient(self) + + +async def test_a_client_starts_an_async_operation_and_waits_later() -> None: + client = FakeClient({"RunJob": RunJobResponse(output="done")}) + + handle = await client.start_run_job(name="nightly") + + # The start gives the handle before the result. + assert client.started == [("RunJob", RunJobRequest(name="nightly"))] + assert client.awaited == [] + assert await handle == RunJobResponse(output="done") + + +async def test_a_client_call_returns_the_result() -> None: + client = FakeClient({"DescribeJob": DescribeJobResponse(status="running")}) + + described = await client.describe_job(name="nightly") + + assert described == DescribeJobResponse(status="running") + assert client.started == [("DescribeJob", DescribeJobRequest(name="nightly"))] + + +async def test_a_call_passes_its_rpc_options_to_the_client() -> None: + client = FakeClient({"RunJob": RunJobResponse(output="done")}) + + handle = await client.start_run_job( + name="nightly", + rpc_metadata={"authorization": "token"}, + rpc_timeout=datetime.timedelta(seconds=5), + ) + _ = await handle + + assert client.rpc_options == [ + ({"authorization": "token"}, datetime.timedelta(seconds=5)) + ] + + +def test_only_an_async_operation_has_a_start_method() -> None: + assert hasattr(JobServiceClientExtension, "start_run_job") + assert not hasattr(JobServiceClientExtension, "start_describe_job") diff --git a/advanced/samples/python/tests/test_client_target.py b/advanced/samples/python/tests/test_client_target.py index 6d4775d1f..bfab4999c 100644 --- a/advanced/samples/python/tests/test_client_target.py +++ b/advanced/samples/python/tests/test_client_target.py @@ -2,6 +2,7 @@ import collections.abc import dataclasses +import datetime import types import typing @@ -124,6 +125,12 @@ class FakeClient( def __init__(self, answers: dict[str, list[object]]) -> None: self.answers: dict[str, list[object]] = answers self.calls: list[Call] = [] + self.rpc_options: list[ + tuple[ + collections.abc.Mapping[str, str | bytes] | None, + datetime.timedelta | None, + ] + ] = [] @override def _create_nexus_client( @@ -131,7 +138,10 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> FakeNexusClient: + self.rpc_options.append((rpc_metadata, rpc_timeout)) return FakeNexusClient(self, service, endpoint) @@ -258,6 +268,35 @@ async def test_a_handle_decodes_its_bodies_with_its_body_type() -> None: assert page.records == [record] +async def test_a_resource_method_passes_its_rpc_options_to_the_client() -> None: + client = FakeClient({"DescribeStream": [DescribeStreamResponse(last_offset=5)]}) + stream = Stream( + body_type=str, + stream=StreamRef(id="s1"), + created=None, + status=None, + _operations=client, + ) + + _ = await stream.describe( + rpc_metadata={"authorization": "token"}, + rpc_timeout=datetime.timedelta(seconds=3), + ) + + assert client.rpc_options == [ + ({"authorization": "token"}, datetime.timedelta(seconds=3)) + ] + + +async def test_workflow_operations_refuse_rpc_options() -> None: + request = DescribeStreamRequest(stream=StreamRef(id="s1")) + + with pytest.raises(ValueError, match="outside a Workflow"): + _ = await WorkflowStreamService()._stream_service_describe_stream( + request, rpc_timeout=datetime.timedelta(seconds=3) + ) + + def test_only_the_client_target_has_the_read_poll() -> None: # The WIT names the client target, since Workflow code reads from a Local Activity. assert hasattr(stream_module, "read_records") diff --git a/advanced/samples/python/tests/test_poll_service.py b/advanced/samples/python/tests/test_poll_service.py index 72265e623..b047d6acf 100644 --- a/advanced/samples/python/tests/test_poll_service.py +++ b/advanced/samples/python/tests/test_poll_service.py @@ -1,6 +1,7 @@ from __future__ import annotations import collections.abc +import datetime import typing import uuid @@ -72,8 +73,10 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> FakeNexusClient: - del service + del service, rpc_metadata, rpc_timeout return FakeNexusClient(self, endpoint) def tokens(self, operation: str) -> list[bytes | None]: diff --git a/advanced/samples/python/wit/client_calls/__init__.py b/advanced/samples/python/wit/client_calls/__init__.py new file mode 100644 index 000000000..4d6263344 --- /dev/null +++ b/advanced/samples/python/wit/client_calls/__init__.py @@ -0,0 +1,57 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import typing + +import nexusrpc +import temporalio.converter +from . import services as _services +from .operations.run_job import run_job +from .operations.describe_job import describe_job + +__all__ = [ + "describe_job", + "run_job", +] + + +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +_SerializationContextFactory = collections.abc.Callable[ + [_InputT], temporalio.converter.SerializationContext +] + + +class _NexusOperationInfo(typing.Generic[_InputT, _OutputT]): + def __init__( + self, + *, + operation: nexusrpc.Operation[_InputT, _OutputT], + serialization_context: _SerializationContextFactory[_InputT] | None = None, + ) -> None: + self.operation: nexusrpc.Operation[_InputT, _OutputT] = operation + self.serialization_context: _SerializationContextFactory[_InputT] | None = ( + serialization_context + ) + + +__nexus_operation_registry__: dict[ + tuple[str, str], _NexusOperationInfo[typing.Any, typing.Any] +] = { + ( + "JobService", + "RunJob", + ): _NexusOperationInfo( + operation=_services.JobService.run_job, + ), + ( + "JobService", + "DescribeJob", + ): _NexusOperationInfo( + operation=_services.JobService.describe_job, + ), +} diff --git a/advanced/samples/python/wit/client_calls/client.py b/advanced/samples/python/wit/client_calls/client.py new file mode 100644 index 000000000..3b3f37431 --- /dev/null +++ b/advanced/samples/python/wit/client_calls/client.py @@ -0,0 +1,181 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import typing +import abc +import datetime + +from .models import ( + DescribeJobRequest, + DescribeJobResponse, + RunJobRequest, + RunJobResponse, +) + + +_OutputT = typing.TypeVar("_OutputT") + + +class _NexusClient(typing.Protocol): + """Starts the Nexus operations of one service.""" + + @typing.overload + async def start_operation( + self, + operation: str, + input: typing.Any, + *, + output_type: type[_OutputT], + ) -> collections.abc.Awaitable[_OutputT]: ... + + @typing.overload + async def start_operation( + self, + operation: str, + input: typing.Any, + ) -> collections.abc.Awaitable[None]: ... + + +class JobServiceClientExtension(abc.ABC): + """Starts the operations of ``JobService`` from code outside a Workflow. + + A client class inherits this class and implements ``_create_nexus_client``. + """ + + def __init_subclass__(cls, **kwargs: typing.Any) -> None: + super().__init_subclass__(**kwargs) + # The MRO picks one base for a name that two bases have, with no error. + for base in cls.__mro__: + if JobServiceClientExtension in base.__mro__: + continue + for name in ( + "run_job", + "start_run_job", + "describe_job", + ): + if name in base.__dict__: + raise TypeError( + f"{cls.__qualname__} gets {name} from {base.__qualname__} and from JobServiceClientExtension" + ) + + @abc.abstractmethod + def _create_nexus_client( + self, + *, + service: str, + endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, + ) -> _NexusClient: + """Return the Nexus client that starts the operations of ``service``. + + ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. + """ + ... + + async def _job_service_run_job( + self, + request: RunJobRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> RunJobResponse: + nexus_client = self._create_nexus_client( + service="JobService", + endpoint="jobs", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + handle = await nexus_client.start_operation( + operation="RunJob", + input=request, + output_type=RunJobResponse, + ) + return await handle + + async def run_job( + self, + *, + name: str, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> RunJobResponse: + """Run a job, which can take a long time.""" + request = RunJobRequest( + name=name, + ) + return await self._job_service_run_job( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) + + async def _job_service_start_run_job( + self, + request: RunJobRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> collections.abc.Awaitable[RunJobResponse]: + nexus_client = self._create_nexus_client( + service="JobService", + endpoint="jobs", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + return await nexus_client.start_operation( + operation="RunJob", + input=request, + output_type=RunJobResponse, + ) + + async def start_run_job( + self, + *, + name: str, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> collections.abc.Awaitable[RunJobResponse]: + """Start ``run_job``, and return its handle. Await the handle for the result.""" + request = RunJobRequest( + name=name, + ) + return await self._job_service_start_run_job( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) + + async def _job_service_describe_job( + self, + request: DescribeJobRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> DescribeJobResponse: + nexus_client = self._create_nexus_client( + service="JobService", + endpoint="jobs", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ) + handle = await nexus_client.start_operation( + operation="DescribeJob", + input=request, + output_type=DescribeJobResponse, + ) + return await handle + + async def describe_job( + self, + *, + name: str, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> DescribeJobResponse: + """Describe a job.""" + request = DescribeJobRequest( + name=name, + ) + return await self._job_service_describe_job( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) diff --git a/advanced/samples/python/wit/client_calls/models.py b/advanced/samples/python/wit/client_calls/models.py new file mode 100644 index 000000000..3c0a15dba --- /dev/null +++ b/advanced/samples/python/wit/client_calls/models.py @@ -0,0 +1,25 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses + + +@dataclasses.dataclass(slots=True) +class RunJobRequest: + name: str + + +@dataclasses.dataclass(slots=True) +class RunJobResponse: + output: str + + +@dataclasses.dataclass(slots=True) +class DescribeJobRequest: + name: str + + +@dataclasses.dataclass(slots=True) +class DescribeJobResponse: + status: str diff --git a/advanced/samples/python/wit/client_calls/operations/__init__.py b/advanced/samples/python/wit/client_calls/operations/__init__.py new file mode 100644 index 000000000..ca23418da --- /dev/null +++ b/advanced/samples/python/wit/client_calls/operations/__init__.py @@ -0,0 +1,3 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations diff --git a/advanced/samples/python/wit/client_calls/operations/describe_job.py b/advanced/samples/python/wit/client_calls/operations/describe_job.py new file mode 100644 index 000000000..0857ff179 --- /dev/null +++ b/advanced/samples/python/wit/client_calls/operations/describe_job.py @@ -0,0 +1,35 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import temporalio.workflow + +from ..models import ( + DescribeJobRequest, + DescribeJobResponse, +) + + +async def _describe_job( + request: DescribeJobRequest, +) -> temporalio.workflow.NexusOperationHandle[DescribeJobResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="JobService", + endpoint="jobs", + ) + return await nexus_client.start_operation( + operation="DescribeJob", + input=request, + output_type=DescribeJobResponse, + ) + + +async def describe_job( + *, + name: str, +) -> temporalio.workflow.NexusOperationHandle[DescribeJobResponse,]: + """Describe a job.""" + request = DescribeJobRequest( + name=name, + ) + return await _describe_job(request) diff --git a/advanced/samples/python/wit/client_calls/operations/run_job.py b/advanced/samples/python/wit/client_calls/operations/run_job.py new file mode 100644 index 000000000..40b9cf37a --- /dev/null +++ b/advanced/samples/python/wit/client_calls/operations/run_job.py @@ -0,0 +1,35 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import temporalio.workflow + +from ..models import ( + RunJobRequest, + RunJobResponse, +) + + +async def _run_job( + request: RunJobRequest, +) -> temporalio.workflow.NexusOperationHandle[RunJobResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="JobService", + endpoint="jobs", + ) + return await nexus_client.start_operation( + operation="RunJob", + input=request, + output_type=RunJobResponse, + ) + + +async def run_job( + *, + name: str, +) -> temporalio.workflow.NexusOperationHandle[RunJobResponse,]: + """Run a job, which can take a long time.""" + request = RunJobRequest( + name=name, + ) + return await _run_job(request) diff --git a/advanced/samples/python/wit/client_calls/services.py b/advanced/samples/python/wit/client_calls/services.py new file mode 100644 index 000000000..f143b59b4 --- /dev/null +++ b/advanced/samples/python/wit/client_calls/services.py @@ -0,0 +1,24 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +from nexusrpc import Operation, service +from .models import ( + DescribeJobRequest, + DescribeJobResponse, + RunJobRequest, + RunJobResponse, +) + + +@service +class JobService: + run_job: Operation[ + RunJobRequest, + RunJobResponse, + ] = Operation(name="RunJob") + + describe_job: Operation[ + DescribeJobRequest, + DescribeJobResponse, + ] = Operation(name="DescribeJob") diff --git a/advanced/samples/python/wit/multi_operation_service/client.py b/advanced/samples/python/wit/multi_operation_service/client.py index 38318edd0..d19d488e2 100644 --- a/advanced/samples/python/wit/multi_operation_service/client.py +++ b/advanced/samples/python/wit/multi_operation_service/client.py @@ -5,6 +5,7 @@ import collections.abc import typing import abc +import datetime from .models import ( Request, @@ -68,20 +69,28 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... async def _interceptor_service_first_operation( self, request: Request, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Response: nexus_client = self._create_nexus_client( service="InterceptorService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="FirstOperation", @@ -94,19 +103,28 @@ async def first_operation( self, *, id: str, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Response: request = Request( id=id, ) - return await self._interceptor_service_first_operation(request) + return await self._interceptor_service_first_operation( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _interceptor_service_second_operation( self, request: Request, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Response: nexus_client = self._create_nexus_client( service="InterceptorService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="SecondOperation", @@ -119,19 +137,28 @@ async def second_operation( self, *, id: str, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Response: request = Request( id=id, ) - return await self._interceptor_service_second_operation(request) + return await self._interceptor_service_second_operation( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _interceptor_service_value_operation( self, request: ValueRequest[ValueT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ValueResponse[ValueT]: nexus_client = self._create_nexus_client( service="InterceptorService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = typing.cast( collections.abc.Awaitable[ValueResponse[ValueT]], @@ -148,9 +175,13 @@ async def value_operation( *, id: str, value: ValueT, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ValueResponse[ValueT]: request = ValueRequest( id=id, value=value, ) - return await self._interceptor_service_value_operation(request) + return await self._interceptor_service_value_operation( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) diff --git a/advanced/samples/python/wit/poll_service/_resources/feed.py b/advanced/samples/python/wit/poll_service/_resources/feed.py index f0a065aea..303a105f3 100644 --- a/advanced/samples/python/wit/poll_service/_resources/feed.py +++ b/advanced/samples/python/wit/poll_service/_resources/feed.py @@ -5,6 +5,7 @@ import collections.abc import dataclasses import typing +import datetime from ..models import ( ListWorkflowsRequest, @@ -33,6 +34,9 @@ def _ops(self) -> _FeedServiceOperations: async def read( self, page_size: int | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: """List one page of the feed.""" request = ListWorkflowsRequest( @@ -40,11 +44,16 @@ async def read( page_size=page_size, query=self.query, ) - return await self._ops()._feed_service_list_feed(request) + return await self._ops()._feed_service_list_feed( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def iter_read( self, page_size: int | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> collections.abc.AsyncIterator[ListWorkflowsResponse]: """Yield each response of ``read``. The poll helper gives the next request, and the iteration stops when it gives ``None``. @@ -55,7 +64,9 @@ async def iter_read( query=self.query, ) while True: - response = await self._ops()._feed_service_list_feed(request) + response = await self._ops()._feed_service_list_feed( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) yield response next_request = _support.next_page_request(request, response) if next_request is None: @@ -69,6 +80,9 @@ class _FeedServiceOperations(typing.Protocol): async def _feed_service_list_feed( self, request: ListWorkflowsRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: ... @@ -81,7 +95,12 @@ def __init__(self, nexus_client: typing.Any) -> None: async def _feed_service_list_feed( self, request: ListWorkflowsRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") nexus_client = self._nexus_client handle = await nexus_client.start_operation( operation="ListFeed", diff --git a/advanced/samples/python/wit/poll_service/client.py b/advanced/samples/python/wit/poll_service/client.py index dca84b459..3c5a16d78 100644 --- a/advanced/samples/python/wit/poll_service/client.py +++ b/advanced/samples/python/wit/poll_service/client.py @@ -6,6 +6,7 @@ import dataclasses import typing import abc +import datetime from .models import ( FeedOrder, @@ -64,20 +65,28 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... async def _page_service_list_workflows( self, request: ListWorkflowsRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: nexus_client = self._create_nexus_client( service="PageService", endpoint="pages", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="ListWorkflows", @@ -93,6 +102,8 @@ async def list_workflows( page_size: int | None = None, next_page_token: bytes | None = None, query: str | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: """List one page of Workflows. The poll iterator lists every page.""" request = ListWorkflowsRequest( @@ -101,7 +112,9 @@ async def list_workflows( next_page_token=next_page_token, query=query, ) - return await self._page_service_list_workflows(request) + return await self._page_service_list_workflows( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def iter_list_workflows( self, @@ -110,6 +123,8 @@ async def iter_list_workflows( page_size: int | None = None, next_page_token: bytes | None = None, query: str | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> collections.abc.AsyncIterator[ListWorkflowsResponse]: """Yield each response of ``list_workflows``. The poll helper gives the next request, and the iteration stops when it gives ``None``. @@ -121,7 +136,9 @@ async def iter_list_workflows( query=query, ) while True: - response = await self._page_service_list_workflows(request) + response = await self._page_service_list_workflows( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) yield response next_request = _support.next_page_request(request, response) if next_request is None: @@ -156,20 +173,28 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... async def _feed_service_open_feed( self, request: OpenFeedRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Feed: nexus_client = self._create_nexus_client( service="FeedService", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="OpenFeed", @@ -185,21 +210,30 @@ async def open_feed( namespace: str, query: str | None = None, order: FeedOrder | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Feed: request = OpenFeedRequest( namespace=namespace, query=query, order=order, ) - return await self._feed_service_open_feed(request) + return await self._feed_service_open_feed( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _feed_service_list_feed( self, request: ListWorkflowsRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: nexus_client = self._create_nexus_client( service="FeedService", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="ListFeed", @@ -215,6 +249,8 @@ async def list_feed( page_size: int | None = None, next_page_token: bytes | None = None, query: str | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ListWorkflowsResponse: request = ListWorkflowsRequest( namespace=namespace, @@ -222,7 +258,9 @@ async def list_feed( next_page_token=next_page_token, query=query, ) - return await self._feed_service_list_feed(request) + return await self._feed_service_list_feed( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def iter_list_feed( self, @@ -231,6 +269,8 @@ async def iter_list_feed( page_size: int | None = None, next_page_token: bytes | None = None, query: str | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> collections.abc.AsyncIterator[ListWorkflowsResponse]: """Yield each response of ``list_feed``. The poll helper gives the next request, and the iteration stops when it gives ``None``. @@ -242,7 +282,9 @@ async def iter_list_feed( query=query, ) while True: - response = await self._feed_service_list_feed(request) + response = await self._feed_service_list_feed( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) yield response next_request = _support.next_page_request(request, response) if next_request is None: diff --git a/advanced/samples/python/wit/stream_service/_resources/stream.py b/advanced/samples/python/wit/stream_service/_resources/stream.py index 4a997b6ab..3ea9e42a4 100644 --- a/advanced/samples/python/wit/stream_service/_resources/stream.py +++ b/advanced/samples/python/wit/stream_service/_resources/stream.py @@ -58,6 +58,9 @@ async def publish( previous_completed_task: WorkflowTaskEpoch | None = None, failed_tasks: list[int] | None = None, withdraw_run: str | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> AppendRecordsResponse: """Publish Records. The answer has the first and the last offset.""" request = AppendRecordsRequest( @@ -69,7 +72,9 @@ async def publish( failed_tasks=failed_tasks or [], withdraw_run=withdraw_run, ) - return await self._ops()._stream_service_append_records(request) + return await self._ops()._stream_service_append_records( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def close( self, @@ -79,6 +84,9 @@ async def close( status: StreamCloseStatus | None = None, previous_completed_task: WorkflowTaskEpoch | None = None, failed_tasks: list[int] | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> None: """Close the Stream. Readers drain what is left, then end.""" request = CloseStreamRequest( @@ -90,7 +98,9 @@ async def close( failed_tasks=failed_tasks or [], status=status, ) - _ = await self._ops()._stream_service_close_stream(request) + _ = await self._ops()._stream_service_close_stream( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def read( self, @@ -99,6 +109,9 @@ async def read( max_records: int | None = None, wait: datetime.timedelta | None = None, create_wait: datetime.timedelta | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ReadRecordsResponse[BodyT]: """Read one page of Records. The poll iterator reads them all.""" request = ReadRecordsRequest( @@ -109,7 +122,9 @@ async def read( wait=wait, create_wait=create_wait, ) - return await self._ops()._stream_service_read_records(request, self.body_type) + return await self._ops()._stream_service_read_records( + request, self.body_type, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def iter_read( self, @@ -118,6 +133,9 @@ async def iter_read( max_records: int | None = None, wait: datetime.timedelta | None = None, create_wait: datetime.timedelta | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> collections.abc.AsyncIterator[ReadRecordsResponse[BodyT]]: """Yield each response of ``read``. The poll helper gives the next request, and the iteration stops when it gives ``None``. @@ -132,7 +150,10 @@ async def iter_read( ) while True: response = await self._ops()._stream_service_read_records( - request, self.body_type + request, + self.body_type, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) yield response next_request = _support.next_read_request(request, response) @@ -142,12 +163,17 @@ async def iter_read( async def describe( self, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> DescribeStreamResponse: """Describe the Stream's status, newest offset and Retention.""" request = DescribeStreamRequest( stream=self.stream, ) - return await self._ops()._stream_service_describe_stream(request) + return await self._ops()._stream_service_describe_stream( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) class _StreamServiceOperations(typing.Protocol): @@ -156,22 +182,34 @@ class _StreamServiceOperations(typing.Protocol): async def _stream_service_append_records( self, request: AppendRecordsRequest[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> AppendRecordsResponse: ... async def _stream_service_close_stream( self, request: CloseStreamRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseStreamResponse: ... async def _stream_service_read_records( self, request: ReadRecordsRequest, body_type: type[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ReadRecordsResponse[BodyT]: ... async def _stream_service_describe_stream( self, request: DescribeStreamRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> DescribeStreamResponse: ... @@ -181,14 +219,24 @@ class WorkflowStreamService: async def _stream_service_append_records( self, request: AppendRecordsRequest[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> AppendRecordsResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") handle = await _append_records(request) return await handle async def _stream_service_close_stream( self, request: CloseStreamRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseStreamResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") handle = await _close_stream(request) return await handle @@ -196,14 +244,24 @@ async def _stream_service_read_records( self, request: ReadRecordsRequest, body_type: type[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ReadRecordsResponse[BodyT]: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") handle = await _read_records(request, body_type) return await handle async def _stream_service_describe_stream( self, request: DescribeStreamRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> DescribeStreamResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") handle = await _describe_stream(request) return await handle diff --git a/advanced/samples/python/wit/stream_service/client.py b/advanced/samples/python/wit/stream_service/client.py index d3699f159..b31229e9f 100644 --- a/advanced/samples/python/wit/stream_service/client.py +++ b/advanced/samples/python/wit/stream_service/client.py @@ -4,8 +4,8 @@ import collections.abc import typing -import datetime import abc +import datetime from .models import ( AppendRecord, @@ -91,10 +91,13 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... @@ -102,10 +105,15 @@ async def _stream_service_create_stream( self, request: CreateStreamRequest, body_type: type[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Stream[BodyT]: nexus_client = self._create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="CreateStream", @@ -132,6 +140,8 @@ async def create_stream( withdraw_run: str | None = None, owner: Producer | None = None, ref: StreamRef | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Stream[BodyT]: """ Args: @@ -153,15 +163,22 @@ async def create_stream( owner=owner, ref=ref, ) - return await self._stream_service_create_stream(request, body_type) + return await self._stream_service_create_stream( + request, body_type, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_service_append_records( self, request: AppendRecordsRequest[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> AppendRecordsResponse: nexus_client = self._create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="AppendRecords", @@ -180,6 +197,8 @@ async def append_records( sequence: int | None = None, previous_completed_task: WorkflowTaskEpoch | None = None, withdraw_run: str | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> AppendRecordsResponse: """Append one batch. Workflow code makes the call from a Local Activity. @@ -203,15 +222,22 @@ async def append_records( previous_completed_task=previous_completed_task, withdraw_run=withdraw_run, ) - return await self._stream_service_append_records(request) + return await self._stream_service_append_records( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_service_close_stream( self, request: CloseStreamRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseStreamResponse: nexus_client = self._create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="CloseStream", @@ -230,6 +256,8 @@ async def close_stream( failure: BaseException | None = None, status: StreamCloseStatus | None = None, previous_completed_task: WorkflowTaskEpoch | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseStreamResponse: """ Args: @@ -248,16 +276,23 @@ async def close_stream( status=status, previous_completed_task=previous_completed_task, ) - return await self._stream_service_close_stream(request) + return await self._stream_service_close_stream( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_service_read_records( self, request: ReadRecordsRequest, body_type: type[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ReadRecordsResponse[BodyT]: nexus_client = self._create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="ReadRecords", @@ -278,6 +313,8 @@ async def read_records( create_wait: datetime.timedelta | None = None, last: int | None = None, max_bytes: int | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> ReadRecordsResponse[BodyT]: """A long poll. NOT_FOUND for an uncreated Stream. Workflow code reads from a Local Activity. @@ -307,7 +344,9 @@ async def read_records( last=last, max_bytes=max_bytes, ) - return await self._stream_service_read_records(request, body_type) + return await self._stream_service_read_records( + request, body_type, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def iter_read_records( self, @@ -321,6 +360,8 @@ async def iter_read_records( create_wait: datetime.timedelta | None = None, last: int | None = None, max_bytes: int | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> collections.abc.AsyncIterator[ReadRecordsResponse[BodyT]]: """Yield each response of ``read_records``. The poll helper gives the next request, and the iteration stops when it gives ``None``. @@ -336,7 +377,9 @@ async def iter_read_records( max_bytes=max_bytes, ) while True: - response = await self._stream_service_read_records(request, body_type) + response = await self._stream_service_read_records( + request, body_type, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) yield response next_request = _support.next_read_request(request, response) if next_request is None: @@ -346,10 +389,15 @@ async def iter_read_records( async def _stream_service_describe_stream( self, request: DescribeStreamRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> DescribeStreamResponse: nexus_client = self._create_nexus_client( service="temporal.api.streamservice.v1.StreamService", endpoint="__temporal_system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="DescribeStream", @@ -362,8 +410,12 @@ async def describe_stream( self, *, stream: StreamRef, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> DescribeStreamResponse: request = DescribeStreamRequest( stream=stream, ) - return await self._stream_service_describe_stream(request) + return await self._stream_service_describe_stream( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) diff --git a/advanced/samples/python/wit/stream_writer/_resources/writer.py b/advanced/samples/python/wit/stream_writer/_resources/writer.py index 4ed798177..3cb7663e6 100644 --- a/advanced/samples/python/wit/stream_writer/_resources/writer.py +++ b/advanced/samples/python/wit/stream_writer/_resources/writer.py @@ -2,8 +2,10 @@ from __future__ import annotations +import collections.abc import dataclasses import typing +import datetime from ..models import ( AppendRecord, @@ -40,31 +42,46 @@ def _ops(self) -> _StreamWriterOperations: async def write_records( self, records: list[AppendRecord[BodyT]] | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WriteRecordsResponse: """Write Records at once. The answer has the first and the last offset.""" request = WriteRecordsRequest( writer_id=self.writer_id, records=records or [], ) - return await self._ops()._stream_writer_write_records(request) + return await self._ops()._stream_writer_write_records( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def close( self, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWriterResponse: """Release the handle. The last one waits for a write, then drops the writer.""" request = CloseWriterRequest( writer_id=self.writer_id, ) - return await self._ops()._stream_writer_close_writer(request) + return await self._ops()._stream_writer_close_writer( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def free( self, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> FreeWriterResponse: """Release the handle with no wait, for a handle that lang drops.""" request = FreeWriterRequest( writer_id=self.writer_id, ) - return await self._ops()._stream_writer_free_writer(request) + return await self._ops()._stream_writer_free_writer( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) class _StreamWriterOperations(typing.Protocol): @@ -73,16 +90,25 @@ class _StreamWriterOperations(typing.Protocol): async def _stream_writer_write_records( self, request: WriteRecordsRequest[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WriteRecordsResponse: ... async def _stream_writer_close_writer( self, request: CloseWriterRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWriterResponse: ... async def _stream_writer_free_writer( self, request: FreeWriterRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> FreeWriterResponse: ... @@ -95,7 +121,12 @@ def __init__(self, nexus_client: typing.Any) -> None: async def _stream_writer_write_records( self, request: WriteRecordsRequest[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WriteRecordsResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") nexus_client = self._nexus_client handle = await nexus_client.start_operation( operation="WriteRecords", @@ -107,7 +138,12 @@ async def _stream_writer_write_records( async def _stream_writer_close_writer( self, request: CloseWriterRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWriterResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") nexus_client = self._nexus_client handle = await nexus_client.start_operation( operation="CloseWriter", @@ -119,7 +155,12 @@ async def _stream_writer_close_writer( async def _stream_writer_free_writer( self, request: FreeWriterRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> FreeWriterResponse: + if rpc_metadata is not None or rpc_timeout is not None: + raise ValueError("RPC options apply to calls outside a Workflow") nexus_client = self._nexus_client handle = await nexus_client.start_operation( operation="FreeWriter", diff --git a/advanced/samples/python/wit/stream_writer/client.py b/advanced/samples/python/wit/stream_writer/client.py index 60a734263..b007197e1 100644 --- a/advanced/samples/python/wit/stream_writer/client.py +++ b/advanced/samples/python/wit/stream_writer/client.py @@ -4,8 +4,8 @@ import collections.abc import typing -import datetime import abc +import datetime from .models import ( AppendRecord, @@ -85,10 +85,13 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... @@ -96,10 +99,15 @@ async def _stream_writer_open_writer( self, request: OpenWriterRequest, body_type: type[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Writer[BodyT]: nexus_client = self._create_nexus_client( service="temporal.sdk.streams.v1.StreamWriter", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="OpenWriter", @@ -125,6 +133,8 @@ async def open_writer( max_batch_size: int | None = None, max_retry_duration: datetime.timedelta | None = None, previous_completed_task: WorkflowTaskEpoch | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> Writer[BodyT]: """ Args: @@ -144,15 +154,22 @@ async def open_writer( max_retry_duration=max_retry_duration, previous_completed_task=previous_completed_task, ) - return await self._stream_writer_open_writer(request, body_type) + return await self._stream_writer_open_writer( + request, body_type, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_writer_write_records( self, request: WriteRecordsRequest[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WriteRecordsResponse: nexus_client = self._create_nexus_client( service="temporal.sdk.streams.v1.StreamWriter", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="WriteRecords", @@ -166,20 +183,29 @@ async def write_records( *, records: list[AppendRecord[BodyT]] | None = None, writer_id: int | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WriteRecordsResponse: request = WriteRecordsRequest( records=records or [], writer_id=writer_id, ) - return await self._stream_writer_write_records(request) + return await self._stream_writer_write_records( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_writer_close_writer( self, request: CloseWriterRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWriterResponse: nexus_client = self._create_nexus_client( service="temporal.sdk.streams.v1.StreamWriter", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="CloseWriter", @@ -192,19 +218,28 @@ async def close_writer( self, *, writer_id: int | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWriterResponse: request = CloseWriterRequest( writer_id=writer_id, ) - return await self._stream_writer_close_writer(request) + return await self._stream_writer_close_writer( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_writer_free_writer( self, request: FreeWriterRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> FreeWriterResponse: nexus_client = self._create_nexus_client( service="temporal.sdk.streams.v1.StreamWriter", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="FreeWriter", @@ -217,19 +252,28 @@ async def free_writer( self, *, writer_id: int | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> FreeWriterResponse: request = FreeWriterRequest( writer_id=writer_id, ) - return await self._stream_writer_free_writer(request) + return await self._stream_writer_free_writer( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) async def _stream_writer_close_writers( self, request: CloseWritersRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWritersResponse: nexus_client = self._create_nexus_client( service="temporal.sdk.streams.v1.StreamWriter", endpoint=None, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="CloseWriters", @@ -240,7 +284,12 @@ async def _stream_writer_close_writers( async def close_writers( self, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> CloseWritersResponse: """Drop every open writer, for example at shutdown.""" request = CloseWritersRequest() - return await self._stream_writer_close_writers(request) + return await self._stream_writer_close_writers( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) diff --git a/advanced/samples/python/wit/workflow_handle/client.py b/advanced/samples/python/wit/workflow_handle/client.py index f395b1d72..c990b4975 100644 --- a/advanced/samples/python/wit/workflow_handle/client.py +++ b/advanced/samples/python/wit/workflow_handle/client.py @@ -4,8 +4,8 @@ import collections.abc import typing -import datetime import abc +import datetime from .models import ( StartWorkflowRequest, @@ -61,20 +61,28 @@ def _create_nexus_client( *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... async def _workflow_handle_service_start_workflow( self, request: StartWorkflowRequest, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WorkflowHandle: nexus_client = self._create_nexus_client( service="WorkflowHandleService", endpoint="temporal-system", + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, ) handle = await nexus_client.start_operation( operation="StartWorkflow", @@ -97,6 +105,8 @@ async def start_workflow( task_queue: str, workflow_start_delay: datetime.timedelta | None = None, namespace: str, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, ) -> WorkflowHandle: """Start a Workflow through System Nexus and get a handle to it.""" request = StartWorkflowRequest( @@ -106,4 +116,6 @@ async def start_workflow( workflow_start_delay=workflow_start_delay, namespace=namespace, ) - return await self._workflow_handle_service_start_workflow(request) + return await self._workflow_handle_service_start_workflow( + request, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) diff --git a/changelog/added/timely-tern.md b/changelog/added/timely-tern.md new file mode 100644 index 000000000..2f7b85d7c --- /dev/null +++ b/changelog/added/timely-tern.md @@ -0,0 +1,2 @@ +Each Python client call takes `rpc_metadata` and `rpc_timeout`, and passes them to `_create_nexus_client`. The methods of a resource take them too, and its Workflow operations raise `ValueError` when either is set. +An operation that WIT declares `async func` also gets the Python client method `start_`, which returns the handle of the operation. diff --git a/src/generator/python.rs b/src/generator/python.rs index 8bbe9a775..563b8bf3a 100644 --- a/src/generator/python.rs +++ b/src/generator/python.rs @@ -859,7 +859,7 @@ impl<'a> ApiPlanner<'a> { ] }) .flatten() - .chain(std::iter::once("abc".to_string())) + .chain(["abc".to_string(), "datetime".to_string()]) .collect::>(); let mut output = String::new(); render_generated_file_header(&mut output); @@ -996,6 +996,9 @@ impl<'a> ApiPlanner<'a> { if targets.workflow && !service.delay_load_temporalio_workflow { module_imports.insert("temporalio.workflow".to_string()); } + if targets.client { + module_imports.insert("datetime".to_string()); + } let mut output = String::new(); render_generated_file_header(&mut output); @@ -1136,7 +1139,7 @@ impl<'a> ApiPlanner<'a> { output.push_str(" async def "); output.push_str(&python_field_name(&method.name)); output.push_str("(\n"); - self.render_resource_method_params(output, method); + self.render_resource_method_params(output, method, operations_object); output.push_str(" ) -> "); output.push_str(&result_annotation); output.push_str(":\n"); @@ -1186,19 +1189,26 @@ impl<'a> ApiPlanner<'a> { } } - fn render_resource_method_params(&self, output: &mut String, method: &PlannedResourceMethod) { + /// The parameters of a resource method. A method of a class that serves the client target + /// also takes the RPC options. + fn render_resource_method_params( + &self, + output: &mut String, + method: &PlannedResourceMethod, + rpc_options: bool, + ) { output.push_str(" self,\n"); // Python refuses a parameter without a default after one with a default, and WIT // allows any order, so such a method takes keyword arguments only. let has_default = |param: &PlannedResourceField| { param.optional || resource_method_list_param(param).is_some() }; - if method + let keyword_only = method .params .iter() .skip_while(|param| !has_default(param)) - .any(|param| !has_default(param)) - { + .any(|param| !has_default(param)); + if keyword_only { output.push_str(" *,\n"); } for param in &method.params { @@ -1222,6 +1232,9 @@ impl<'a> ApiPlanner<'a> { } output.push_str(",\n"); } + if rpc_options { + render_rpc_option_parameters(output, " ", keyword_only); + } } /// The iterator of a method whose operation polls. The class serves every generated target, @@ -1260,7 +1273,7 @@ impl<'a> ApiPlanner<'a> { output.push_str("\n async def "); output.push_str(&poll_function_name(&method_name)); output.push_str("(\n"); - self.render_resource_method_params(output, method); + self.render_resource_method_params(output, method, targets.client); output.push_str(" ) -> collections.abc.AsyncIterator["); if targets.workflow { output.push_str(&temporalio_workflow_type_annotation( @@ -1283,7 +1296,7 @@ impl<'a> ApiPlanner<'a> { let type_args = operation.runtime_type_args("self."); let response = if targets.client { vec![format!( - "response = await self._ops().{}(request{type_args})", + "response = await self._ops().{}(request{type_args}, {PYTHON_RPC_OPTION_ARGS})", client_operation_name(service, operation) )] } else if service.endpoint.is_none() { @@ -1700,6 +1713,7 @@ impl<'a> ApiPlanner<'a> { .runtime_type_definition .value() .map(str::to_string), + asynchronous: operation.asynchronous, }) } @@ -2971,6 +2985,8 @@ struct RenderedOperation<'a> { output_record_type_parameters: Vec, /// The definition type that a public call takes in place of the runtime types. runtime_type_definition: Option, + /// Whether WIT declares the operation `async`. + asynchronous: bool, } impl RenderedOperation<'_> { @@ -5866,15 +5882,19 @@ fn client_extension_class_name(service: &RenderedService<'_>) -> String { /// Each public method of the package's client extensions belongs to one service, so one client /// class can inherit them all. fn validate_client_names(services: &[RenderedService<'_>]) -> Result<()> { - let mut owners = BTreeMap::<&str, &str>::new(); + let mut owners = BTreeMap::::new(); for service in services { for operation in &service.operations { - if let Some(first_service) = owners.insert(&operation.attr_name, service.name) { - return Err(Error::ClientTargetNameConflict { - name: operation.attr_name.clone(), - first_service: first_service.to_string(), - second_service: service.name.to_string(), - }); + let start_name = operation_has_start_method(operation) + .then(|| start_function_name(&operation.attr_name)); + for name in std::iter::once(operation.attr_name.clone()).chain(start_name) { + if let Some(first_service) = owners.insert(name.clone(), service.name) { + return Err(Error::ClientTargetNameConflict { + name, + first_service: first_service.to_string(), + second_service: service.name.to_string(), + }); + } } } } @@ -5913,14 +5933,80 @@ const PYTHON_CLIENT_EXTENSION_HOOK: &str = r#" *, service: str, endpoint: str | None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None, + rpc_timeout: datetime.timedelta | None, ) -> _NexusClient: """Return the Nexus client that starts the operations of ``service``. ``endpoint`` is ``None`` for a service with no endpoint, which the client serves. + The RPC options are those of one call. """ ... "#; +/// The RPC options of a call outside a Workflow, with their annotations. +const PYTHON_RPC_OPTIONS: [(&str, &str); 2] = [ + ( + "rpc_metadata", + "collections.abc.Mapping[str, str | bytes] | None", + ), + ("rpc_timeout", "datetime.timedelta | None"), +]; + +const PYTHON_RPC_OPTION_ARGS: &str = "rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout"; + +/// The keyword parameters of the RPC options, one line each at `indent`. `keyword_only` says +/// whether the parameters before them already end in a `*` marker. +fn render_rpc_option_parameters(output: &mut String, indent: &str, keyword_only: bool) { + render_rpc_option_parameters_with_default(output, indent, keyword_only, "None"); +} + +fn render_rpc_option_parameters_with_default( + output: &mut String, + indent: &str, + keyword_only: bool, + default: &str, +) { + if !keyword_only { + output.push_str(indent); + output.push_str("*,\n"); + } + for (name, annotation) in PYTHON_RPC_OPTIONS { + output.push_str(indent); + output.push_str(name); + output.push_str(": "); + output.push_str(annotation); + output.push_str(" = "); + output.push_str(default); + output.push_str(",\n"); + } +} + +/// The arguments of a call that an operation function makes with its request. +fn render_request_call_arguments( + output: &mut String, + target: PythonTarget, + operation: &RenderedOperation<'_>, +) { + output.push_str("(request"); + output.push_str(&operation.runtime_type_args("")); + if target == PythonTarget::Client { + output.push_str(", "); + output.push_str(PYTHON_RPC_OPTION_ARGS); + } + output.push_str(")\n"); +} + +/// Whether a client starts the operation with `start_`, which returns the handle. Only an +/// operation that WIT declares `async` can run for a long time. +fn operation_has_start_method(operation: &RenderedOperation<'_>) -> bool { + operation.asynchronous && !operation_returns_direct(operation) +} + +fn start_function_name(name: &str) -> String { + format!("start_{name}") +} + fn render_client_extension(output: &mut String, service: &RenderedService<'_>) { let class_name = client_extension_class_name(service); output.push_str("class "); @@ -5954,6 +6040,12 @@ fn render_client_extension(output: &mut String, service: &RenderedService<'_>) { for operation in &service.operations { output.push_str(&python_string_literal(&operation.attr_name)); output.push_str(", "); + if operation_has_start_method(operation) { + output.push_str(&python_string_literal(&start_function_name( + &operation.attr_name, + ))); + output.push_str(", "); + } } output.push_str("):\n"); output.push_str(" if name in base.__dict__:\n"); @@ -6327,6 +6419,8 @@ fn render_resource_method_operations_body( output.push_str(&client_operation_name(service, operation)); output.push_str("(request"); output.push_str(&operation.runtime_type_args("self.")); + output.push_str(", "); + output.push_str(PYTHON_RPC_OPTION_ARGS); output.push_str(")\n"); } @@ -6391,6 +6485,10 @@ fn render_workflow_operations_class( let result_annotation = render_operations_method_signature(output, service, operation, true); output.push('\n'); + output.push_str(" if rpc_metadata is not None or rpc_timeout is not None:\n"); + output.push_str( + " raise ValueError(\"RPC options apply to calls outside a Workflow\")\n", + ); if service.endpoint.is_none() { render_inline_operation_call( output, @@ -6436,6 +6534,7 @@ fn render_operations_method_signature( output.push_str(",\n"); } operation.render_runtime_type_parameters(output, " "); + render_rpc_option_parameters(output, " ", false); output.push_str(" ) -> "); output.push_str(&result_annotation); output.push(':'); @@ -7376,9 +7475,18 @@ fn render_function_unpacked_overload( } } - if !primary_case.is_some_and(|function_case| !function_case.positional_args.is_empty()) { + let keyword_marker = + !primary_case.is_some_and(|function_case| !function_case.positional_args.is_empty()); + if keyword_marker { start_line(output, target, " *,\n"); } + let keyword_only = keyword_marker + || primary_case.is_some_and(|function_case| { + function_case + .positional_args + .iter() + .any(|positional_arg| positional_arg.variadic) + }); for field in &unpacked_input.parameters { if primary_function.is_some_and(|function| function.callable_field_name == field.attr_name) @@ -7455,6 +7563,9 @@ fn render_function_unpacked_overload( } output.push_str(",\n"); } + if target == PythonTarget::Client { + render_rpc_option_parameters_with_default(output, " ", keyword_only, "..."); + } start_line(output, target, ") -> "); output.push_str(&function_unpacked_overload_return_annotation( @@ -7849,6 +7960,153 @@ fn render_operation_functions( output.push('\n'); render_operation_poll_function(output, target, service, operation, next_expr); } + if target == PythonTarget::Client && operation_has_start_method(operation) { + output.push('\n'); + render_client_start_methods(output, service, operation); + } +} + +/// The methods that start an operation and return its handle, with no wait for the result. +fn render_client_start_methods( + output: &mut String, + service: &RenderedService<'_>, + operation: &RenderedOperation<'_>, +) { + let target = PythonTarget::Client; + let private_name = format!( + "_{}_{}", + service.name.to_snake_case(), + start_function_name(&operation.attr_name) + ); + let handle_annotation = format!( + "collections.abc.Awaitable[{}]", + operation_result_annotation(operation) + ); + render_operation_signature_start(output, target, &private_name); + if let Some(input) = &operation.input { + start_line(output, target, " request: "); + output.push_str(&input.annotation); + output.push_str(",\n"); + } + operation.render_runtime_type_parameters(output, " "); + render_rpc_option_parameters(output, " ", false); + start_line(output, target, ") -> "); + output.push_str(&handle_annotation); + output.push_str(":\n"); + render_inline_nexus_client(output, target, service); + if operation.generic_model_types { + start_line(output, target, " return typing.cast(\n"); + start_line(output, target, " "); + output.push_str(&handle_annotation); + output.push_str(",\n"); + start_line( + output, + target, + " await nexus_client.start_operation(\n", + ); + } else { + start_line( + output, + target, + " return await nexus_client.start_operation(\n", + ); + } + let indent = if operation.generic_model_types { + " " + } else { + " " + }; + start_line(output, target, indent); + output.push_str("operation="); + output.push_str(&python_string_literal(operation.wire_name)); + output.push_str(",\n"); + start_line(output, target, indent); + output.push_str("input="); + output.push_str(operation_input_wire_expr(operation)); + output.push_str(",\n"); + if !operation.output_none { + start_line(output, target, indent); + output.push_str("output_type="); + output.push_str(&operation.output_type_expr_for("")); + output.push_str(",\n"); + } + if operation.generic_model_types { + start_line(output, target, " ),\n"); + } + start_line(output, target, " )\n"); + + output.push('\n'); + let public_name = start_function_name(&operation.attr_name); + render_operation_signature_start(output, target, &public_name); + operation.render_runtime_type_parameters(output, " "); + let unpacked_input = operation + .target_unpacked_input(target) + .filter(|unpacked_input| unpacked_input.functions.is_empty()); + if let Some(unpacked_input) = unpacked_input { + if !unpacked_input.parameters.is_empty() { + start_line(output, target, " *,\n"); + } + for field in &unpacked_input.parameters { + start_line(output, target, " "); + 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, target.body_indent()); + } + output.push_str(",\n"); + } + render_rpc_option_parameters(output, " ", !unpacked_input.parameters.is_empty()); + } else { + if operation.input.is_some() { + start_line(output, target, " request: "); + output.push_str(operation_input_annotation(operation)); + output.push_str(",\n"); + } + render_rpc_option_parameters(output, " ", false); + } + start_line(output, target, ") -> "); + output.push_str(&handle_annotation); + output.push_str(":\n"); + render_python_docstring( + output, + target.body_indent(), + Some(&format!( + "Start ``{}``, and return its handle. Await the handle for the result.", + operation.attr_name + )), + &[], + None, + false, + ); + if let Some(unpacked_input) = unpacked_input { + for flattened_message in &unpacked_input.flattened_messages { + render_flattened_message_setup(output, target.body_indent(), flattened_message); + } + start_line(output, target, " request = "); + output.push_str(&unpacked_input.model_name); + output.push_str("(\n"); + for field in &unpacked_input.request_fields { + start_line(output, target, " "); + output.push_str(&field.attr_name); + output.push('='); + output.push_str(&unpacked_request_field_value_expr(field)); + output.push_str(",\n"); + } + start_line(output, target, " )\n"); + } + start_line(output, target, " return await self."); + output.push_str(&private_name); + if operation.input.is_some() { + render_request_call_arguments(output, target, operation); + } else { + output.push('('); + output.push_str(PYTHON_RPC_OPTION_ARGS); + output.push_str(")\n"); + } } /// The name of the iterator of a polled operation or method. @@ -7897,6 +8155,10 @@ fn render_operation_poll_function( output.push_str(operation_input_annotation(operation)); output.push_str(",\n"); } + if target == PythonTarget::Client { + let keyword_only = unpacked_input.is_some_and(|input| !input.parameters.is_empty()); + render_rpc_option_parameters(output, " ", keyword_only); + } start_line(output, target, ") -> collections.abc.AsyncIterator["); match target { PythonTarget::Workflow => output.push_str(&temporalio_workflow_type_annotation( @@ -7931,8 +8193,13 @@ fn render_operation_poll_function( } start_line(output, target, " )\n"); } + let rpc_args = if target == PythonTarget::Client { + format!(", {PYTHON_RPC_OPTION_ARGS}") + } else { + String::new() + }; let call = format!( - "{}(request{})", + "{}(request{}{rpc_args})", operation_call_target(target, service, operation), operation.runtime_type_args("") ); @@ -8031,6 +8298,7 @@ fn render_client_operation_method( output.push_str(",\n"); } operation.render_runtime_type_parameters(output, " "); + render_rpc_option_parameters(output, " ", false); start_line(output, target, ") -> "); output.push_str(&operation_result_annotation(operation)); output.push_str(":\n"); @@ -8129,6 +8397,7 @@ fn render_client_request_method( output.push_str(",\n"); } operation.render_public_runtime_type_parameters(output, " "); + render_rpc_option_parameters(output, " ", false); start_line(output, target, ") -> "); output.push_str(&operation_result_annotation(operation)); output.push_str(":\n"); @@ -8136,13 +8405,14 @@ fn render_client_request_method( operation.render_definition_type_arguments(output, target.body_indent()); start_line(output, target, " return await self."); output.push_str(&client_operation_name(service, operation)); + output.push('('); if operation.input.is_some() { - output.push_str("(request"); + output.push_str("request"); output.push_str(&operation.runtime_type_args("")); - output.push_str(")\n"); - } else { - output.push_str("()\n"); + output.push_str(", "); } + output.push_str(PYTHON_RPC_OPTION_ARGS); + output.push_str(")\n"); } fn function_unpacked_needs_overloads(functions: &[RenderedFunctionField]) -> bool { @@ -8776,6 +9046,9 @@ fn render_unpacked_operation_function( } output.push_str(",\n"); } + if target == PythonTarget::Client { + render_rpc_option_parameters(output, " ", !unpacked_input.parameters.is_empty()); + } render_unpacked_return_annotation(output, target, service, operation); render_unpacked_operation_docstring(output, body_indent, operation, unpacked_input, None); operation.render_definition_type_arguments(output, body_indent); @@ -8795,9 +9068,7 @@ fn render_unpacked_operation_function( start_line(output, target, " )\n"); start_line(output, target, " return await "); output.push_str(&operation_call_target(target, service, operation)); - output.push_str("(request"); - output.push_str(&operation.runtime_type_args("")); - output.push_str(")\n"); + render_request_call_arguments(output, target, operation); } /// The return annotation of an unpacked operation. A Workflow function returns the operation @@ -8921,14 +9192,18 @@ fn render_function_unpacked_implementation( .parameters .iter() .any(|field| rendered_function_implementation_renders_field(field, primary_function)); - if primary_function.is_none() + let keyword_marker = primary_function.is_none() || (primary_function .and_then(python_function_fixed_positional_args) .is_some() - && has_keyword_only_parameters) - { + && has_keyword_only_parameters); + if keyword_marker { start_line(output, target, " *,\n"); } + // A variadic primary function takes `*positional_args`, which is a marker too. + let keyword_only = keyword_marker + || primary_function + .is_some_and(|function| python_function_fixed_positional_args(function).is_none()); for field in &unpacked_input.parameters { if !rendered_function_implementation_renders_field(field, primary_function) { @@ -8970,6 +9245,9 @@ fn render_function_unpacked_implementation( } output.push_str(",\n"); } + if target == PythonTarget::Client { + render_rpc_option_parameters(output, " ", keyword_only); + } render_unpacked_return_annotation(output, target, service, operation); render_unpacked_operation_docstring( @@ -9068,9 +9346,7 @@ fn render_function_unpacked_implementation( start_line(output, target, " )\n"); start_line(output, target, " return await "); output.push_str(&operation_call_target(target, service, operation)); - output.push_str("(request"); - output.push_str(&operation.runtime_type_args("")); - output.push_str(")\n"); + render_request_call_arguments(output, target, operation); } fn rendered_function_implementation_renders_field( @@ -9188,6 +9464,10 @@ fn render_inline_nexus_client( start_line(output, target, " endpoint="); output.push_str(&endpoint); output.push_str(",\n"); + if target == PythonTarget::Client { + start_line(output, target, " rpc_metadata=rpc_metadata,\n"); + start_line(output, target, " rpc_timeout=rpc_timeout,\n"); + } start_line(output, target, " )\n"); } diff --git a/src/parser/json_schema.rs b/src/parser/json_schema.rs index bf6c5ab6d..73cebaa06 100644 --- a/src/parser/json_schema.rs +++ b/src/parser/json_schema.rs @@ -6186,6 +6186,7 @@ fn build_operation( serialization_context: LanguageStringSpec::default(), poll_next: LanguageStringSpec::default(), poll_target: None, + asynchronous: false, data: (), }) } diff --git a/src/parser/wit.rs b/src/parser/wit.rs index 617164a7f..964376518 100644 --- a/src/parser/wit.rs +++ b/src/parser/wit.rs @@ -3117,6 +3117,7 @@ fn build_operation( serialization_context, poll_next, poll_target, + asynchronous: matches!(function.kind, FunctionKind::AsyncFreestanding), data: (), }) } diff --git a/src/planning/type_planning.rs b/src/planning/type_planning.rs index 9163a77fc..90d239b20 100644 --- a/src/planning/type_planning.rs +++ b/src/planning/type_planning.rs @@ -705,6 +705,7 @@ impl<'a> TypePlanningContext<'a> { serialization_context: operation.serialization_context.clone(), poll_next: operation.poll_next.clone(), poll_target: operation.poll_target, + asynchronous: operation.asynchronous, data: PlannedOperationData { output_resource_return: plan_operation_resource_return( &self.spec, @@ -1988,6 +1989,7 @@ interface alerts { serialization_context: LanguageStringSpec::default(), poll_next: LanguageStringSpec::default(), poll_target: None, + asynchronous: false, data: (), }], resources: Vec::new(), diff --git a/src/spec.rs b/src/spec.rs index 89075508f..c4acbee08 100644 --- a/src/spec.rs +++ b/src/spec.rs @@ -674,6 +674,8 @@ pub struct OperationSpec { pub poll_next: F::Text, /// The only target whose API gets the poll. Without it, every target gets the poll. pub poll_target: Option, + /// Whether the operation can run for a long time, so a client can start it and wait later. + pub asynchronous: bool, pub data: F::OperationData, } @@ -712,6 +714,7 @@ impl OperationSpec { serialization_context: map.map_text(self.serialization_context), poll_next: map.map_text(self.poll_next), poll_target: self.poll_target, + asynchronous: self.asynchronous, data, } }