fix(samples): install the stream sample's Stream protos into temporalio

advanced/samples/python/pyproject.toml

@@ -8,13 +8,19 @@ dependencies = [
     "pytest-asyncio>=1.2.0",
     "ruff>=0.15.12",
     "temporalio>=1.34.0",
+    "temporalio-stream-protos",
     "typing_extensions>=4.15.0",
 ]
 
 [tool.uv]
 package = false
 exclude-newer = "2 weeks"
 
+[tool.uv.sources]
+# Keep it non-editable. An editable install adds a path entry, and Python ignores that path for
+# modules of the installed temporalio package.
+temporalio-stream-protos = { path = "temporalio_stream_protos" }
+
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 pythonpath = ["."]
@@ -23,11 +29,10 @@ pythonpath = ["."]
 line-length = 88
 target-version = "py310"
 # The protoc output of the stream protos.
-extend-exclude = ["streamprotos"]
+extend-exclude = ["temporalio_stream_protos"]
 
 [tool.basedpyright]
-# The stream service imports Stream protos that temporalio doesn't have yet.
-exclude = [".venv", "streamprotos", "wit/stream_service"]
+exclude = [".venv", "temporalio_stream_protos"]
 reportAny = false
 reportExplicitAny = false
 reportPrivateUsage = false

advanced/samples/python/temporalio_stream_protos/pyproject.toml

@@ -0,0 +1,30 @@
+# The Python modules of the Stream protos of temporalio/api#895, at the paths where temporalio
+# will have them. Drop this package once a temporalio release has them.
+#
+# Python and basedpyright find these modules only inside the installed temporalio package,
+# since temporalio is a regular package. So they can't be files of the sample project. A local
+# temporalio/ directory hides the installed package, or Python skips it.
+[project]
+name = "temporalio-stream-protos"
+version = "0.1.0"
+requires-python = ">=3.10"
+
+# uv installs a path dependency from a wheel that it builds, so the package needs a backend.
+[build-system]
+requires = ["uv_build>=0.11,<0.12"]
+build-backend = "uv_build"
+
+# temporalio/ here has no __init__.py, so the wheel adds the Stream modules to the installed
+# temporalio and replaces none of its files.
+[tool.uv.build-backend]
+module-name = "temporalio"
+module-root = ""
+namespace = true
+
+# A change to a module rebuilds the wheel, not only a change to this file.
+[tool.uv]
+cache-keys = [
+    { file = "pyproject.toml" },
+    { file = "temporalio/**/*.py" },
+    { file = "temporalio/**/*.pyi" },
+]

advanced/samples/python/streamprotos/temporal/api/stream/v1/message_pb2.pyi → advanced/samples/python/temporalio_stream_protos/temporalio/api/stream/v1/message_pb2.pyi

@@ -9,7 +9,7 @@ from google.protobuf import message as _message
 from google.protobuf import timestamp_pb2 as _timestamp_pb2
 from google.protobuf.internal import containers as _containers
 from temporalio.api.common.v1 import message_pb2 as _message_pb2
-from streamprotos.temporal.api.enums.v1 import stream_pb2 as _stream_pb2
+from temporalio.api.enums.v1 import stream_pb2 as _stream_pb2
 from temporalio.api.failure.v1 import message_pb2 as _message_pb2_1
 import builtins as _builtins
 import sys

advanced/samples/python/streamprotos/temporal/api/streamservice/v1/request_response_pb2.pyi → advanced/samples/python/temporalio_stream_protos/temporalio/api/streamservice/v1/request_response_pb2.pyi

@@ -9,9 +9,9 @@ from google.protobuf import duration_pb2 as _duration_pb2
 from google.protobuf import message as _message
 from google.protobuf.internal import containers as _containers
 from temporalio.api.common.v1 import message_pb2 as _message_pb2_1
-from streamprotos.temporal.api.enums.v1 import stream_pb2 as _stream_pb2
+from temporalio.api.enums.v1 import stream_pb2 as _stream_pb2
 from temporalio.api.failure.v1 import message_pb2 as _message_pb2_2
-from streamprotos.temporal.api.stream.v1 import message_pb2 as _message_pb2
+from temporalio.api.stream.v1 import message_pb2 as _message_pb2
 import builtins as _builtins
 import sys
 import typing as _typing

advanced/samples/python/tests/test_stream_service.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import pytest
+from temporalio.api.stream.v1 import message_pb2
+from temporalio.api.streamservice.v1 import request_response_pb2
+from temporalio.converter import PayloadConverter
+import temporalio.nexus.system
+
+from wit.stream_service import models
+from wit.stream_service.models import (
+    AppendRecordsRequest,
+    ReadRecordsResponse,
+    StreamClose,
+    StreamCloseStatus,
+    StreamData,
+    StreamItemClose,
+    StreamItemData,
+    StreamItemSuperseded,
+    StreamRecord,
+    StreamRef,
+    Supersession,
+)
+
+
+@pytest.fixture(autouse=True)
+def user_payload_converter(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        temporalio.nexus.system,
+        "_current_user_payload_converter",
+        lambda: PayloadConverter.default,
+    )
+
+
+def test_read_items_round_trip_as_their_records() -> None:
+    converter = models._ReadRecordsResponseTransferTypeConverter()
+    model = ReadRecordsResponse(
+        records=[
+            StreamRecord(
+                item=StreamItemData(StreamData(body={"text": "hi"})),
+                offset=1,
+            ),
+            StreamRecord(
+                item=StreamItemSuperseded(
+                    Supersession(producer_id="workflow-1", failed_tasks=[7])
+                ),
+                offset=2,
+            ),
+            StreamRecord(
+                item=StreamItemClose(
+                    StreamClose(status=StreamCloseStatus.StreamCloseStatusCompleted)
+                ),
+                offset=3,
+            ),
+        ],
+        next_offset=4,
+        done=True,
+    )
+
+    wire = converter.to_transfer_type(model)
+
+    # The wire messages come from `temporalio-stream-protos`, since no temporalio release has them.
+    assert isinstance(wire, request_response_pb2.ReadRecordsResponse)
+    assert all(isinstance(record, message_pb2.StreamRecord) for record in wire.records)
+    assert [record.WhichOneof("item") for record in wire.records] == [
+        "data",
+        "superseded",
+        "close",
+    ]
+    assert wire.records[1].superseded.failed_tasks == [7]
+    assert converter.from_transfer_type(wire, ReadRecordsResponse) == model
+
+
+def test_a_record_of_an_unknown_item_kind_has_no_item() -> None:
+    converter = models._StreamRecordTransferTypeConverter()
+    # Offset 1, and an item in field 5, which only a newer server knows.
+    wire = message_pb2.StreamRecord.FromString(b"\x08\x01\x2a\x00")
+
+    decoded = converter.from_transfer_type(wire, StreamRecord)
+
+    assert decoded == StreamRecord(item=None, offset=1)
+
+
+def test_a_request_without_its_stream_is_refused() -> None:
+    converter = models._AppendRecordsRequestTransferTypeConverter()
+    wire = converter.to_transfer_type(AppendRecordsRequest(stream=StreamRef(id="s-1")))
+    assert wire.stream.id == "s-1"
+
+    wire.ClearField("stream")
+    with pytest.raises(
+        ValueError, match="missing required field AppendRecordsRequest.stream"
+    ):
+        _ = converter.from_transfer_type(wire, AppendRecordsRequest)

changelog/changed/nesting-newt.md

@@ -0,0 +1 @@
+The `stream-service` sample's Python project installs the Stream protos of `temporalio/api#895` into `temporalio` from a local `temporalio-stream-protos` package, so basedpyright checks the stream service.