[Brook] Added the stream handles over Core and refused unsupported owners.

changelog/added/pirouetting-platypus.md

@@ -0,0 +1 @@
+Add experimental `temporalio.contrib.streams` so Activities and clients can append ordered records to a Workflow's stream in Redis, and outside readers can follow it from a cursor.

temporalio/contrib/streams/__init__.py

@@ -34,7 +34,22 @@
 
 A store is configured as a plugin:
 ``Client.connect(..., plugins=[RedisStreams(...)])`` registers it on the
-client and on every Worker built from that client.
+client and on every Worker built from that client. Each
+context then reaches a stream the same way:
+
+- An Activity reaches the stream of the Workflow that scheduled it with
+  :func:`activity_handle`, and writes as itself: its Activity id with the
+  scheduling run id, and its Temporal attempt.
+- Any process holding a client reaches a Workflow's stream with
+  :func:`get_stream_handle`, and writes with a producer id and an attempt of
+  its own.
+
+Every read follows the Workflow's run chain across Continue-as-New and ends
+when the chain closes. A reader can't pin one run.
+
+In this release only a Workflow owns a stream, and only Activities and
+clients read one. Reading inside a Workflow (:func:`workflow_reader`) and
+the other owner kinds raise :class:`StreamUnsupportedError`.
 
 The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in
 :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body``
@@ -55,6 +70,13 @@
     StreamStorageError,
     StreamUnsupportedError,
 )
+from temporalio.contrib.streams._handles import (
+    ActivityStreamHandle,
+    StreamHandle,
+    StreamProducer,
+    activity_handle,
+    get_stream_handle,
+)
 from temporalio.contrib.streams._plugin import StreamStorePlugin
 from temporalio.contrib.streams._record import (
     Cursor,
@@ -64,8 +86,10 @@
 )
 from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef
 from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic
+from temporalio.contrib.streams._workflow import workflow_reader
 
 __all__ = [
+    "ActivityStreamHandle",
     "BEGINNING",
     "Cursor",
     "DEFAULT_TOPIC",
@@ -75,9 +99,11 @@
     "StreamCursorError",
     "StreamError",
     "StreamExpiredError",
+    "StreamHandle",
     "StreamNotFoundError",
     "StreamOutcomeUnknownError",
     "StreamOwnerKind",
+    "StreamProducer",
     "StreamProducerError",
     "StreamRecord",
     "StreamRecordError",
@@ -88,5 +114,8 @@
     "StreamTopic",
     "StreamUnsupportedError",
     "Supersession",
+    "activity_handle",
+    "get_stream_handle",
     "topic",
+    "workflow_reader",
 ]