feat(python): return a client call's result, with its RPC options

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_<name>`, 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_<name>`.
+
 ```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.
 

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

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;
+}

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"]

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"]

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);
+}

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")

advanced/samples/python/tests/test_client_target.py

@@ -2,6 +2,7 @@
 
 import collections.abc
 import dataclasses
+import datetime
 import types
 import typing
 
@@ -124,14 +125,23 @@ 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(
         self,
         *,
         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")

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]:

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,
+    ),
+}