[Brook S5] Added the client Stream and Stream reads.

temporalio/client/__init__.py

@@ -185,6 +185,7 @@
     ScheduleUpdate,
     ScheduleUpdateInput,
 )
+from ._streams import Stream
 from ._worker_versioning import (
     BuildIdOp,
     BuildIdOpAddNewCompatible,
@@ -397,4 +398,5 @@
     "ParamType",
     "ReturnType",
     "SelfType",
+    "Stream",
 ]

temporalio/client/_client.py

@@ -13,13 +13,14 @@
     TYPE_CHECKING,
     Any,
     Concatenate,
+    TypeVar,
     cast,
     overload,
 )
 
 import nexusrpc
 from nexusrpc import OutputT
-from typing_extensions import Required, Self, TypedDict
+from typing_extensions import Required, Self, TypedDict, Unpack
 
 import temporalio.activity
 import temporalio.api.workflowservice.v1
@@ -110,10 +111,19 @@
 )
 
 if TYPE_CHECKING:
-    from temporalio.common import StreamBackend
+    from temporalio.common import (
+        Creating,
+        StreamBackend,
+        StreamDefinition,
+        StreamHandle,
+        StreamOptions,
+    )
 
     from ._interceptor import Interceptor
     from ._plugin import Plugin
+    from ._streams import Stream
+
+_StreamPayloadT = TypeVar("_StreamPayloadT")
 
 
 class Client:
@@ -911,6 +921,102 @@ def get_workflow_handle(
             result_type=result_type,
         )
 
+    def create_stream(
+        self,
+        definition: type[StreamDefinition[_StreamPayloadT]],
+        *,
+        id: str,
+        backend: str | None = None,
+        **options: Unpack[StreamOptions],
+    ) -> Creating[Stream[_StreamPayloadT]]:
+        """Create a Stream that the client owns, or get the one that exists.
+
+        A create is create-or-get by Stream id. On a Stream that exists, it ignores ``options``
+        and returns that Stream. A Stream must exist before its first publish. The client closes
+        a Stream that it owns, and a Workflow only writes to it. Await the call to get the
+        Stream, or use it in ``async with``, which closes the Stream at the end of the block::
+
+            async with client.create_stream(Turns, id="chat-42") as stream:
+                await stream.publish(Turn("user", "hi"))
+
+        .. warning::
+            This API is experimental.
+
+        Args:
+            definition: The Stream definition class.
+            id: The Stream id, which the caller picks, as for a Workflow id.
+            backend: The name of the backend. Defaults to the registered backend with
+                ``default=True``.
+            options: The options of a new Stream, such as ``retention``. See
+                :py:class:`temporalio.common.StreamOptions`.
+
+        Raises:
+            StreamConfigError: The client has no backend with that name, or no backend has
+                ``default=True`` and the call gives no backend. No call is made.
+            StreamClosedError: A Stream of that id was created and closed before.
+            StreamExpiredError: A Stream of that id expired.
+        """
+        from temporalio.common._streams import _stream_client
+
+        return _stream_client.create(
+            self, definition, id=id, backend=backend, options=options
+        )
+
+    def get_stream(
+        self,
+        definition: type[StreamDefinition[_StreamPayloadT]],
+        *,
+        id: str,
+        backend: str | None = None,
+    ) -> Stream[_StreamPayloadT]:
+        """Get a Stream by its id. It makes no call.
+
+        To create a Stream that the client owns, use :py:meth:`create_stream`. For a handle
+        that came as data, such as from a Workflow's query, use :py:meth:`to_stream`.
+
+        .. warning::
+            This API is experimental.
+
+        Args:
+            definition: The Stream definition class. Reads decode each Record with its payload
+                type.
+            id: The Stream id.
+            backend: The name of the backend. Defaults to the registered backend with
+                ``default=True``.
+
+        Raises:
+            StreamConfigError: The client has no backend with that name, or no backend has
+                ``default=True`` and the call gives no backend.
+            ValueError: The id is empty.
+        """
+        from temporalio.common._streams import _stream_client
+
+        return _stream_client.get_stream(self, definition, id=id, backend=backend)
+
+    def to_stream(
+        self,
+        handle: StreamHandle[_StreamPayloadT],
+        *,
+        payload_type: type[_StreamPayloadT] | None = None,
+    ) -> Stream[_StreamPayloadT]:
+        """Get the Stream of a handle that came as data, such as from a query. It makes no call.
+
+        .. warning::
+            This API is experimental.
+
+        Args:
+            handle: The handle of the Stream.
+            payload_type: The type that reads decode each Record with. Defaults to no type, so
+                a read decodes each Record as a JSON value.
+
+        Raises:
+            StreamConfigError: The client has no backend of the handle, or the handle has no
+                backend and no backend has ``default=True``.
+        """
+        from temporalio.common._streams import _stream_client
+
+        return _stream_client.to_stream(self, handle, payload_type=payload_type)
+
     def get_workflow_handle_for(
         self,
         workflow: (

temporalio/client/_streams.py

@@ -0,0 +1,202 @@
+"""Streams outside Workflow code.
+
+This module imports the Streams API of :py:mod:`temporalio.common` on first use, so a client
+with no Stream never imports it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Mapping
+from datetime import timedelta
+from types import TracebackType
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
+
+from typing_extensions import Self
+
+if TYPE_CHECKING:
+    from temporalio.common import End, StreamHandle, Subscription
+    from temporalio.common._streams._stream_client import StreamCalls
+
+T = TypeVar("T")
+
+
+class Stream(Generic[T]):
+    """A Stream outside Workflow code, with its calls. ``T`` is the payload type.
+
+    Get one with :py:meth:`temporalio.client.Client.create_stream`,
+    :py:meth:`temporalio.client.Client.get_stream`, or
+    :py:meth:`temporalio.client.Client.to_stream` for a handle that came as data. Its
+    :py:attr:`handle` is the data to pass on, such as to a Workflow.
+
+    All Streams of one client and Stream id write as one Producer, and :py:meth:`producer`
+    returns a Stream that writes as a named Producer. A write error that no caller got raises at
+    the next call of the Stream.
+
+    .. warning::
+        This API is experimental.
+    """
+
+    def __init__(
+        self, handle: StreamHandle[T], payload_type: Any, calls: StreamCalls
+    ) -> None:
+        """Create a Stream. Use a client's ``create_stream``, ``get_stream`` or ``to_stream``."""
+        self._handle = handle
+        # The type that a read decodes each value with. None decodes with no type hint.
+        self._payload_type = payload_type
+        self._calls = calls
+
+    @property
+    def handle(self) -> StreamHandle[T]:
+        """The data that names the Stream, to pass on."""
+        return self._handle
+
+    def publish(
+        self,
+        *values: T,
+        rpc_metadata: Mapping[str, str | bytes] = {},
+        rpc_timeout: timedelta | None = None,
+    ) -> asyncio.Future[int]:
+        """Start the write of one Record for each value, in order, and return its future.
+
+        The write has its place in the Stream's order when the call returns, also when no
+        caller awaits the future. The writes of one Stream keep the order of the calls. A cancel
+        of the future doesn't stop the write. A list is one value. The next call of the Stream
+        raises a write error that no caller got.
+
+        Args:
+            values: One value for each Record.
+            rpc_metadata: Headers of each call. The Redis and memory backends make no RPC, so
+                they send none.
+            rpc_timeout: The deadline of each call. A call past it raises ``RPCError`` with
+                ``DEADLINE_EXCEEDED``.
+
+        Returns:
+            The future of the write. It gives the offset of the last Record, or raises
+            ``StreamError`` when the backend refused the write, such as
+            ``StreamNotFoundError`` for a Stream that nobody created.
+
+        Raises:
+            StreamClosedError: This object closed the Stream.
+        """
+        return self._calls.publish(values, rpc_metadata, rpc_timeout)
+
+    def subscribe(
+        self,
+        *,
+        offset: int | End | None = None,
+        last: int | None = None,
+        wait_new_records: bool = True,
+        create_wait: timedelta | None = None,
+        rpc_metadata: Mapping[str, str | bytes] = {},
+        rpc_timeout: timedelta | None = None,
+    ) -> Subscription[T]:
+        """Read the Stream's items in order.
+
+        The read ends after a ``Closed`` item, or at the newest Record when
+        ``wait_new_records`` is False. To stop after a quiet period, use your own timer, and
+        resume later with ``offset=`` set to the read's ``next_offset``. A ``Superseded`` item
+        withdraws earlier Records of a failed attempt.
+
+        Args:
+            offset: Start at this offset, inclusive, or at ``common.END`` for new
+                Records only. Defaults to the first Record.
+            last: Start with the newest ``last`` Records, oldest first, such as for the first
+                view of a chat. Can't be used with ``offset``. A Record that a later
+                ``Superseded`` item withdraws doesn't count. At most 1000.
+            wait_new_records: Wait for new Records at the end of the Stream.
+            create_wait: How long to wait for a Stream that doesn't exist yet, such as the
+                Stream of a Workflow that just started. Defaults to no wait.
+            rpc_metadata: Headers of each call of the read. See :py:meth:`publish`.
+            rpc_timeout: The deadline of each call of the read, not of the whole read. A call
+                that waits for new Records returns before the deadline.
+        """
+        from temporalio.common._streams._access import read_spec
+
+        read = read_spec(
+            self._payload_type,
+            offset=offset,
+            last=last,
+            wait_new_records=wait_new_records,
+            create_wait=create_wait,
+        )
+        return self._calls.subscribe(read, rpc_metadata, rpc_timeout)
+
+    async def get_offset(
+        self,
+        *,
+        rpc_metadata: Mapping[str, str | bytes] = {},
+        rpc_timeout: timedelta | None = None,
+    ) -> int:
+        """The next offset of the Stream, the newest offset plus one, or 0 if it has no Record.
+
+        Offsets are sparse and above zero, so ``offset=0`` reads from the start.
+        ``rpc_metadata`` and ``rpc_timeout`` are as for :py:meth:`publish`.
+        """
+        return await self._calls.get_offset(rpc_metadata, rpc_timeout)
+
+    async def close(
+        self,
+        *,
+        result: Any = None,
+        failure: BaseException | None = None,
+        rpc_metadata: Mapping[str, str | bytes] = {},
+        rpc_timeout: timedelta | None = None,
+    ) -> None:
+        """Write the Records that this object published, then close the Stream.
+
+        Readers get the Records before the close, then a ``Closed`` item with ``result`` or
+        ``failure``. A cancel failure, such as a ``CancelledError``, closes the Stream as
+        canceled. ``rpc_metadata`` and ``rpc_timeout`` are as for :py:meth:`publish`.
+
+        Raises:
+            StreamNotOwnerError: The code doesn't own the Stream, such as a client with a
+                Workflow's Stream. Only the run chain closes a Workflow's Stream, and only a
+                client closes a client's Stream. A client with a Workflow's Stream raises with
+                no call.
+        """
+        await self._calls.close(result, failure, rpc_metadata, rpc_timeout)
+
+    async def __aenter__(self) -> Self:
+        """Use the Stream in ``async with``, which closes it at the end.
+
+        Raises:
+            StreamNotOwnerError: The code can't close the Stream. See :py:meth:`close`.
+        """
+        await self._calls.enter()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        """Close the Stream, with the exception as its failure if there is one.
+
+        A cancel closes it with a ``CancelledError`` failure. Another ``BaseException``, such as
+        ``KeyboardInterrupt``, leaves it open. If this object already closed the Stream, this
+        does nothing.
+        """
+        await self._calls.exit(exc)
+
+    def producer(self, name: str, epoch: int) -> Stream[T]:
+        """Return this Stream as a Stream that writes as the Producer ``name`` at ``epoch``.
+
+        Use it for a writer that must fence its earlier instances, such as a server that
+        restarts. Keep ``epoch`` in durable storage, and raise it at each start. After a write
+        at a higher epoch, a write at a lower one raises
+        :py:class:`temporalio.common.StreamFencedError`. The Records of the earlier epochs aren't
+        withdrawn, since they're real messages. Run only one live writer for each name.
+
+        Args:
+            name: The Producer name, the same across restarts.
+            epoch: The writer's epoch, 1 or more.
+        """
+        return Stream(
+            self._handle, self._payload_type, self._calls.producer(name, epoch)
+        )
+
+    def __repr__(self) -> str:
+        """Show the handle."""
+        return f"{type(self).__name__}({self._handle!r})"

temporalio/common/__init__.py

@@ -63,6 +63,7 @@
         Superseded,
     )
     from temporalio.common._streams._plugin import StreamBackend
+    from temporalio.common._streams._reader import Subscription
     from temporalio.common._streams.system.models import (
         Producer,
         ProducerEpoch,
@@ -132,6 +133,7 @@
     "StreamSuperseded",
     "StreamTooLargeError",
     "StreamUnsupportedError",
+    "Subscription",
     "Superseded",
     "TypedSearchAttributes",
     "VersioningBehavior",
@@ -1626,6 +1628,7 @@ def wait_sync(self, timeout: float | None = None) -> None:
         "StreamSuperseded",
         "StreamTooLargeError",
         "StreamUnsupportedError",
+        "Subscription",
         "Superseded",
         "WorkflowTaskEpoch",
         "defn",

temporalio/common/_streams/__init__.py

@@ -25,6 +25,7 @@
     Superseded,
 )
 from temporalio.common._streams._plugin import StreamBackend
+from temporalio.common._streams._reader import Subscription
 
 # nexgen generates these wire models. Records, Superseded items and errors carry them.
 from temporalio.common._streams.system.models import (
@@ -64,6 +65,7 @@
     "StreamSuperseded",
     "StreamTooLargeError",
     "StreamUnsupportedError",
+    "Subscription",
     "Superseded",
     "WorkflowTaskEpoch",
     "defn",