feat(python): add a resource factory, and pass a resource as data

GUIDE.md

@@ -22,6 +22,7 @@ of `@nexus` directives.
   - [Resource Declaration](#resource-declaration)
   - [Name-Binding: How Resource Methods Find Operations](#name-binding-how-resource-methods-find-operations)
   - [Resource Return Binding](#resource-return-binding)
+  - [Resource Factories](#resource-factories)
   - [Edge Cases and Error Conditions](#edge-cases-and-error-conditions)
 - [Proto-Backed Models](#proto-backed-models)
   - [Transfer-Type Conversion](#transfer-type-conversion)
@@ -741,6 +742,47 @@ result = await handle  # CreateJobResponse
 return CreateJobResult(job=Job(job_id=result.job_id), created=result.created)
 ```
 
+### Resource Factories
+
+Python gives each resource a factory, which builds a handle on an existing
+resource with no call. The client extension has it as a method, and the Workflow
+package has it as a function. It takes the definition of a resource with
+`@nexus.definition`, or its runtime types, and then the constructor fields by
+keyword. The client's handle calls through that client, and a Workflow's handle
+calls the operations of the code that runs it.
+
+`@nexus.flatten-in-api` on the resource makes the factory take some fields of
+one constructor parameter's record in place of the parameter. With `keep=true`,
+the factory also takes the whole value, and a call gives either the value or the
+required promoted fields.
+
+```wit
+/// @nexus.flatten-in-api param="stream" fields="id backend" keep=true
+resource %stream {
+  constructor(%stream: stream-ref, created: option<bool>, status: option<stream-status>);
+}
+```
+
+```python
+stream = client.stream(Turns, id="chat-1:turns", backend="redis")
+same = client.stream(Turns, stream=StreamRef(id="chat-1:turns", backend="redis"))
+```
+
+The factory has no sibling fields, so it takes no `fill`. It must promote every
+required field of the record. The other refusals of a field's flatten apply too.
+The factory's name must not be the name of another client method, or of another
+function or module of the Workflow package. An optional resource field defaults
+to `None`. A dataclass with a required field after an optional one takes its
+fields by keyword.
+
+**A resource as data.** A resource with one required field of a proto-backed
+record, and no other required field, gets a transfer converter. Its wire form is
+that record, so a handle can be an argument or a result. In Python the parameter
+type gives the type arguments of a decoded handle, as in `Stream[Turn]`, and the
+handle takes the operations of the code that runs it. TypeScript, .NET and Go
+keep their constructors, and .NET passes a handle in the JSON form of its
+properties.
+
 ### Edge Cases and Error Conditions
 
 #### Same field name in constructor and method parameter
@@ -2218,15 +2260,20 @@ The referenced helper must be available from the package specified by
 
 ### @nexus.flatten-in-api
 
-**Placement:** Record type (with `@nexus.proto`), or a field whose type is such
-a record **Syntax:** `@nexus.flatten-in-api` (no arguments) on a record,
+**Placement:** Record type (with `@nexus.proto`), a field whose type is such a
+record, or a resource **Syntax:** `@nexus.flatten-in-api` (no arguments) on a
+record,
 `@nexus.flatten-in-api fields="<field> ..." fill="<field>=<sibling> ..."` on a
-field
+field,
+`@nexus.flatten-in-api param="<parameter>" fields="<field> ..." keep=true` on a
+resource
 
 Flattens the record's fields into the parent record's generated API surface
 instead of creating a nested model. On a field, it flattens only the fields that
 `fields` names, and `fill` copies sibling fields into the record. See
-[Flattening some fields of one field](#flattening-some-fields-of-one-field).
+[Flattening some fields of one field](#flattening-some-fields-of-one-field). On
+a resource, it flattens one constructor parameter in the factory. See
+[Resource Factories](#resource-factories).
 
 ```wit
 /// @nexus.proto "temporal.api.sdk.v1.UserMetadata"

advanced/samples/README.md

@@ -195,7 +195,9 @@ Local Activity. The Stream and the writer are generic in the record body type
 create of a Stream takes a definition, which is a type of the sample support
 code, since no SDK has `StreamDefinition` yet. The requests take a source for
 each target, which the sample support code gives in place of Brook's runtime.
-The create takes the backend of the Stream's ref, and the ref copies the id.
+The create takes the backend of the Stream's ref, and the ref copies the id. In
+Python, the factory `stream` builds a handle from an id or a ref with no call,
+and a handle passes as its ref.
 
 - WIT: [`inputs/stream-service.wit`](inputs/stream-service.wit),
   [`inputs/stream-writer.wit`](inputs/stream-writer.wit)

advanced/samples/inputs/stream-service.wit

@@ -378,6 +378,7 @@ interface stream-service {
   /// @nexus.definition python="temporal_support.StreamDefinition"
   ///   typescript="StreamDefinition" typescript-import="../../support/temporal_model_converters"
   ///   dotnet="Nexgen.Support.StreamDefinition"
+  /// @nexus.flatten-in-api param="stream" fields="id backend" keep=true
   resource %stream {
     constructor(%stream: stream-ref, created: option<bool>, status: option<stream-status>);
 

advanced/samples/invalid/client-target/factory-name-clash/expected.toml

@@ -0,0 +1,6 @@
+languages = ["python"]
+targets = ["client"]
+
+[error]
+kind = "client-target-name-conflict"
+contains = ["`start_job` is a client method of both"]

advanced/samples/invalid/client-target/factory-name-clash/input.wit

@@ -0,0 +1,21 @@
+package test:factory-name-clash@1.0.0;
+
+world system { export job-service; }
+
+/// @nexus.endpoint "jobs"
+interface job-service {
+  // The client's factory of `start-job` has the name of the `start_job` of `job`.
+  resource start-job {
+    constructor(id: string);
+  }
+
+  record job-request {
+    id: string,
+  }
+
+  record job-response {
+    id: string,
+  }
+
+  job: async func(request: job-request) -> job-response;
+}

advanced/samples/invalid/flatten/resource-fill/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["on a resource, takes `param`, `fields` and `keep`, not `fill`"]

advanced/samples/invalid/flatten/resource-fill/input.wit

@@ -0,0 +1,17 @@
+package test:resource-fill@1.0.0;
+
+world system { export history-service; }
+
+/// @nexus.endpoint "history"
+interface history-service {
+  record workflow-execution {
+    workflow-id: string,
+    run-id: option<string>,
+  }
+
+  // A factory has no sibling fields, so it takes no fill.
+  /// @nexus.flatten-in-api param="execution" fields="workflow-id" fill="run-id=workflow-id"
+  resource run {
+    constructor(execution: workflow-execution);
+  }
+}

advanced/samples/invalid/flatten/resource-keep-not-a-bool/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["`keep=yes` isn't `true` or `false`"]

advanced/samples/invalid/flatten/resource-keep-not-a-bool/input.wit

@@ -0,0 +1,17 @@
+package test:resource-keep-not-a-bool@1.0.0;
+
+world system { export history-service; }
+
+/// @nexus.endpoint "history"
+interface history-service {
+  record workflow-execution {
+    workflow-id: string,
+    run-id: option<string>,
+  }
+
+  // `keep` is `true` or `false`.
+  /// @nexus.flatten-in-api param="execution" fields="workflow-id" keep=yes
+  resource run {
+    constructor(execution: workflow-execution);
+  }
+}

advanced/samples/invalid/flatten/resource-name-clash/expected.toml

@@ -0,0 +1,5 @@
+descriptors = "temporal"
+
+[error]
+kind = "invalid-wit"
+contains = ["promotes `run_id`, whose name `run-id` is also the name of parameter `run-id`"]

advanced/samples/invalid/flatten/resource-name-clash/input.wit

@@ -0,0 +1,18 @@
+package temporal:nexus@1.0.0;
+
+world system { export history-service; }
+
+/// @nexus.endpoint "history"
+interface history-service {
+  /// @nexus.proto "temporal.api.common.v1.WorkflowExecution"
+  record workflow-execution {
+    workflow-id: string,
+    run-id: option<string>,
+  }
+
+  // The promoted `run-id` has the name of another parameter.
+  /// @nexus.flatten-in-api param="execution" fields="workflow-id run-id"
+  resource run {
+    constructor(execution: workflow-execution, run-id: option<string>);
+  }
+}

advanced/samples/invalid/flatten/resource-not-proto-backed/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit"
+contains = ["resource `run` parameter `execution` flattens fields", "isn't a proto-backed record"]

advanced/samples/invalid/flatten/resource-not-proto-backed/input.wit

@@ -0,0 +1,17 @@
+package test:resource-not-proto-backed@1.0.0;
+
+world system { export history-service; }
+
+/// @nexus.endpoint "history"
+interface history-service {
+  record workflow-execution {
+    workflow-id: string,
+    run-id: option<string>,
+  }
+
+  // The factory can't flatten a record that no proto backs.
+  /// @nexus.flatten-in-api param="execution" fields="workflow-id"
+  resource run {
+    constructor(execution: workflow-execution);
+  }
+}

advanced/samples/invalid/flatten/resource-required-field-not-promoted/expected.toml

@@ -0,0 +1,5 @@
+descriptors = "temporal"
+
+[error]
+kind = "invalid-wit"
+contains = ["doesn't promote its required field `workflow_id`"]

advanced/samples/invalid/flatten/resource-required-field-not-promoted/input.wit

@@ -0,0 +1,18 @@
+package temporal:nexus@1.0.0;
+
+world system { export history-service; }
+
+/// @nexus.endpoint "history"
+interface history-service {
+  /// @nexus.proto "temporal.api.common.v1.WorkflowExecution"
+  record workflow-execution {
+    workflow-id: string,
+    run-id: option<string>,
+  }
+
+  // The factory has no fill, so it must take the required `workflow-id`.
+  /// @nexus.flatten-in-api param="execution" fields="run-id"
+  resource run {
+    constructor(execution: workflow-execution);
+  }
+}

advanced/samples/invalid/flatten/resource-unknown-parameter/expected.toml

@@ -0,0 +1,3 @@
+[error]
+kind = "invalid-wit-directive"
+contains = ["`ref` isn't a parameter of the constructor"]

advanced/samples/invalid/flatten/resource-unknown-parameter/input.wit

@@ -0,0 +1,17 @@
+package test:resource-unknown-parameter@1.0.0;
+
+world system { export history-service; }
+
+/// @nexus.endpoint "history"
+interface history-service {
+  record workflow-execution {
+    workflow-id: string,
+    run-id: option<string>,
+  }
+
+  // `ref` isn't a parameter of the constructor.
+  /// @nexus.flatten-in-api param="ref" fields="workflow-id"
+  resource run {
+    constructor(execution: workflow-execution);
+  }
+}

advanced/samples/invalid/name-collisions/python-workflow-factory/expected.toml

@@ -0,0 +1,5 @@
+languages = ["python"]
+
+[error]
+kind = "python-generated-name-conflict"
+contains = ["the factory of resource `Models`", "the module `models` of the package"]

advanced/samples/invalid/name-collisions/python-workflow-factory/input.wit

@@ -0,0 +1,17 @@
+package test:python-workflow-factory@1.0.0;
+
+world system { export job-service; }
+
+/// @nexus.endpoint "jobs"
+interface job-service {
+  // The Workflow factory of `models` has the name of the module of the models.
+  resource models {
+    constructor(id: string);
+  }
+
+  record job-request {
+    id: string,
+  }
+
+  job: func(request: job-request);
+}

advanced/samples/python/tests/test_client_target.py

@@ -203,6 +203,45 @@ async def test_a_create_takes_the_ref_fields_as_parameters() -> None:
     assert (wire.ref.id, wire.ref.backend) == ("s1", "redis")
 
 
+async def test_a_factory_builds_a_handle_with_no_call() -> None:
+    client = FakeClient({"DescribeStream": [DescribeStreamResponse(last_offset=5)]})
+
+    by_id = client.stream(Turns, id="s1", backend="redis")
+    by_ref = client.stream(Turns, stream=StreamRef(id="s2"))
+
+    _ = assert_type(by_id, Stream[Turn])
+    assert by_id == Stream(body_type=Turn, stream=StreamRef(id="s1", backend="redis"))
+    assert by_id.body_type is Turn
+    assert by_ref.stream == StreamRef(id="s2")
+    assert client.calls == []
+    # The handle calls through the client that built it.
+    _ = await by_id.describe()
+    assert [call.operation for call in client.calls] == ["DescribeStream"]
+
+
+def test_a_factory_takes_the_ref_fields_or_the_ref() -> None:
+    client = FakeClient({})
+
+    with pytest.raises(TypeError, match="pass id or stream"):
+        _ = client.stream(Turns)
+    with pytest.raises(TypeError, match="not both"):
+        _ = client.stream(Turns, id="s1", stream=StreamRef(id="s1"))
+
+
+async def test_a_handle_passes_as_its_ref() -> None:
+    converter = temporalio.converter.DataConverter.default
+    stream = Stream(body_type=Turn, stream=StreamRef(id="s1"), created=True)
+
+    (payload,) = await converter.encode([stream])
+    (decoded,) = await converter.decode([payload], [Stream[Turn]])
+
+    # The wire form is the ref, and the parameter type gives the body type. Only a create's
+    # handle has `created`.
+    assert payload.metadata["messageType"] == b"temporal.api.stream.v1.StreamRef"
+    assert decoded == Stream(body_type=Turn, stream=StreamRef(id="s1"))
+    assert decoded.body_type is Turn
+
+
 async def test_a_client_call_returns_the_result() -> None:
     client = FakeClient({"DescribeStream": [DescribeStreamResponse(last_offset=5)]})
 

advanced/samples/python/tests/test_stream_sources.py

@@ -9,6 +9,9 @@
 import temporalio.workflow
 
 import temporal_support
+from typing_extensions import assert_type
+import wit.stream_service as stream_service
+from wit.stream_service._resources import Stream
 from wit.stream_service.models import (
     CreateStreamRequest,
     CreateStreamResult,
@@ -71,3 +74,14 @@ def create_nexus_client(**_: object) -> FakeNexusClient:
         )
     ]
     assert stream.body_type is str
+
+
+def test_a_workflow_factory_builds_a_handle_on_the_run_operations() -> None:
+    handle = stream_service.stream(Turns, id="chat-1:r1:turns", backend="redis")
+
+    _ = assert_type(handle, Stream[str])
+    assert handle == Stream(
+        body_type=str, stream=StreamRef(id="chat-1:r1:turns", backend="redis")
+    )
+    # A handle with no operations of its own calls those of the code that runs it.
+    assert handle._operations is None

advanced/samples/python/wit/poll_service/_resources/feed.py

@@ -19,7 +19,7 @@ class Feed:
     """The Workflows of one query."""
 
     namespace: str
-    query: str | None
+    query: str | None = None
     _operations: _FeedServiceOperations | None = dataclasses.field(
         default=None, repr=False, compare=False, kw_only=True
     )

advanced/samples/python/wit/poll_service/client.py

@@ -159,6 +159,7 @@ def __init_subclass__(cls, **kwargs: typing.Any) -> None:
             if FeedServiceClientExtension in base.__mro__:
                 continue
             for name in (
+                "feed",
                 "open_feed",
                 "list_feed",
             ):
@@ -290,3 +291,16 @@ async def iter_list_feed(
             if next_request is None:
                 return
             request = next_request
+
+    def feed(
+        self,
+        *,
+        namespace: str,
+        query: str | None = None,
+    ) -> Feed:
+        """A handle on an existing ``Feed``, with no call."""
+        return Feed(
+            namespace=namespace,
+            query=query,
+            _operations=self,
+        )