[Brook S3] Added Stream refs, Records and errors in temporalio.streams.

temporalio/streams/__init__.py

@@ -1 +1,84 @@
-"""The Streams API, which Workflows, Activities and clients share."""
+"""The Streams API, which Workflows, Activities and clients share.
+
+.. warning::
+    The Streams API is experimental.
+"""
+
+from temporalio.streams._errors import (
+    StreamAlreadyExistsError,
+    StreamClosedError,
+    StreamConfigError,
+    StreamError,
+    StreamExpiredError,
+    StreamFencedError,
+    StreamNotFoundError,
+    StreamNotOwnerError,
+    StreamOffsetTooHighError,
+    StreamSeqDivergentError,
+    StreamStorageUnavailableError,
+    StreamTooLargeError,
+    StreamUnsupportedError,
+)
+from temporalio.streams._items import END, End, ItemKind, Record, Superseded
+from temporalio.streams._ref import StreamOptions, StreamRef
+
+# nexgen generates these wire models. Refs, Records, Superseded items and errors carry them.
+from temporalio.streams.system.models import (
+    ActivityEpoch,
+    ClientEpoch,
+    Execution,
+    ExecutionType,
+    LocalActivityEpoch,
+    Producer,
+    ProducerEpoch,
+    ProducerEpochKind,
+    ProducerEpochKindActivity,
+    ProducerEpochKindClient,
+    ProducerEpochKindLocalActivity,
+    ProducerEpochKindStandaloneActivity,
+    ProducerEpochKindWorkflow,
+    StreamErrorReason,
+    StreamSuperseded,
+    Timestamp,
+    WorkflowTaskEpoch,
+)
+
+__all__ = [
+    "END",
+    "ActivityEpoch",
+    "ClientEpoch",
+    "End",
+    "Execution",
+    "ExecutionType",
+    "ItemKind",
+    "LocalActivityEpoch",
+    "Producer",
+    "ProducerEpoch",
+    "ProducerEpochKind",
+    "ProducerEpochKindActivity",
+    "ProducerEpochKindClient",
+    "ProducerEpochKindLocalActivity",
+    "ProducerEpochKindStandaloneActivity",
+    "ProducerEpochKindWorkflow",
+    "Record",
+    "StreamAlreadyExistsError",
+    "StreamClosedError",
+    "StreamConfigError",
+    "StreamError",
+    "StreamErrorReason",
+    "StreamExpiredError",
+    "StreamFencedError",
+    "StreamNotFoundError",
+    "StreamNotOwnerError",
+    "StreamOffsetTooHighError",
+    "StreamOptions",
+    "StreamRef",
+    "StreamSeqDivergentError",
+    "StreamStorageUnavailableError",
+    "StreamSuperseded",
+    "StreamTooLargeError",
+    "StreamUnsupportedError",
+    "Superseded",
+    "Timestamp",
+    "WorkflowTaskEpoch",
+]

temporalio/streams/_errors.py

@@ -0,0 +1,94 @@
+"""Errors of the Streams API."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import temporalio.api.stream.v1
+import temporalio.nexus.system._support as _support
+from temporalio.streams.system import models
+
+# The generated classes are the Stream errors, one for each reason that has its own. A reason
+# with no class of its own, such as STORAGE_CORRUPT, raises the base class.
+from temporalio.streams.system.errors import (
+    StreamAlreadyExistsError,
+    StreamClosedError,
+    StreamError,
+    StreamExpiredError,
+    StreamFencedError,
+    StreamNotFoundError,
+    StreamNotOwnerError,
+    StreamOffsetTooHighError,
+    StreamSeqDivergentError,
+    StreamStorageUnavailableError,
+    StreamTooLargeError,
+    StreamUnsupportedError,
+    stream_error,
+)
+
+__all__ = [
+    "StreamAlreadyExistsError",
+    "StreamClosedError",
+    "StreamConfigError",
+    "StreamError",
+    "StreamExpiredError",
+    "StreamFencedError",
+    "StreamNotFoundError",
+    "StreamNotOwnerError",
+    "StreamOffsetTooHighError",
+    "StreamSeqDivergentError",
+    "StreamStorageUnavailableError",
+    "StreamTooLargeError",
+    "StreamUnsupportedError",
+]
+
+
+class StreamConfigError(RuntimeError):
+    """The Stream backend config is wrong, such as when no backend has a given name.
+
+    It isn't a :py:class:`StreamError`, so ``except StreamError`` doesn't catch it. In Workflow
+    code it fails the Workflow Task. An Activity that raises it retries.
+
+    .. warning::
+        This API is experimental.
+    """
+
+
+def no_default_backend(
+    owner: str, names: Sequence[str], hint: str
+) -> StreamConfigError:
+    """The error of a call with no backend name when no backend has ``default=True``."""
+    return StreamConfigError(
+        f"No Stream backend of the {owner} ({', '.join(names)}) has default=True, so the "
+        "default is native, which this SDK does not have. Mark a backend with "
+        f"default=True, or give its name in {hint}."
+    )
+
+
+def from_reason(reason: int, message: str) -> StreamError:
+    """The Stream error of a ``StreamErrorReason`` number."""
+    try:
+        known = models.StreamErrorReason(reason)
+    except ValueError:
+        # A reason that this SDK doesn't know gives the base class.
+        known = models.StreamErrorReason.StreamErrorReasonUnspecified
+    return stream_error(models.StreamErrorDetails(reason=known or None), message)
+
+
+def from_call_error(err: Exception) -> StreamError:
+    """Turn a failed call of the crate's runtime into the Stream error for its reason.
+
+    ``err.args`` are the gRPC code, the message and the encoded ``StreamErrorDetails``.
+    """
+    _, message, details = err.args
+    proto = temporalio.api.stream.v1.StreamErrorDetails.FromString(details)
+    return from_reason(proto.reason, message)
+
+
+def from_activity_error(err: Exception) -> Exception:
+    """The Stream error that a write Local Activity raised, else ``err``."""
+    found = _support.error_details(err, models.StreamErrorDetails)
+    if found is None:
+        return err
+    details, message = found
+    return stream_error(details, message)

temporalio/streams/_ids.py

@@ -0,0 +1,44 @@
+"""Producer ids. They must match the temporalio-streams crate byte for byte."""
+
+from __future__ import annotations
+
+import urllib.parse
+
+
+def encode_part(text: str) -> str:
+    """Percent-encode every byte but the unreserved ones."""
+    return urllib.parse.quote(text, safe="")
+
+
+def _join(*parts: str) -> str:
+    return ":".join(encode_part(part) for part in parts)
+
+
+def workflow_producer_id(run_id: str) -> str:
+    """Producer id of a Workflow run."""
+    return f"wf:{_join(run_id)}"
+
+
+def local_activity_producer_id(run_id: str, call_key: str) -> str:
+    """Producer id of a Local Activity call."""
+    return f"la:{_join(run_id, call_key)}"
+
+
+def activity_producer_id(run_id: str, activity_id: str, schedule_key: str) -> str:
+    """Producer id of an Activity that a Workflow scheduled."""
+    return f"act:{_join(run_id, activity_id, schedule_key)}"
+
+
+def standalone_activity_producer_id(activity_id: str, activity_run_id: str) -> str:
+    """Producer id of a Standalone Activity."""
+    return f"sa:{_join(activity_id, activity_run_id)}"
+
+
+def client_producer_id(identity: str, process_key: str) -> str:
+    """Producer id of a client process."""
+    return f"client:{_join(identity, process_key)}"
+
+
+def named_client_producer_id(name: str) -> str:
+    """Producer id of a client with a name of its own, which it keeps across restarts."""
+    return f"client:{_join(name)}"