diff --git a/temporalio/contrib/streams/__init__.py b/temporalio/contrib/streams/__init__.py index f365744292..f4e1a8c065 100644 --- a/temporalio/contrib/streams/__init__.py +++ b/temporalio/contrib/streams/__init__.py @@ -21,6 +21,9 @@ refused with :class:`StreamCursorError`, and one whose record retention dropped with :class:`StreamExpiredError`. A read with no cursor starts at :data:`BEGINNING` or :data:`END`. +3. **Topics are defined once.** :func:`topic` defines a topic with the type + its records decode to, and every party shares that definition. A call + that names no topic addresses :data:`DEFAULT_TOPIC`. The stored record is ``temporal.sdk.streams.v1.StreamRecord``, in :mod:`temporalio.bridge.proto.streams.v1`, with the user's value in ``body`` @@ -47,10 +50,13 @@ StreamRecord, Supersession, ) +from temporalio.contrib.streams._ref import StreamOwnerKind, StreamRef +from temporalio.contrib.streams._topic import DEFAULT_TOPIC, StreamTopic, topic __all__ = [ "BEGINNING", "Cursor", + "DEFAULT_TOPIC", "END", "RecordKind", "StreamClosedError", @@ -59,11 +65,15 @@ "StreamExpiredError", "StreamNotFoundError", "StreamOutcomeUnknownError", + "StreamOwnerKind", "StreamProducerError", "StreamRecord", "StreamRecordError", + "StreamRef", "StreamRefusedError", "StreamStorageError", + "StreamTopic", "StreamUnsupportedError", "Supersession", + "topic", ] diff --git a/temporalio/contrib/streams/_ref.py b/temporalio/contrib/streams/_ref.py new file mode 100644 index 0000000000..e41c7cb7d8 --- /dev/null +++ b/temporalio/contrib/streams/_ref.py @@ -0,0 +1,95 @@ +"""A reference to one stream that crosses a process boundary as data. + +A handle is bound to a client and its store, so it cannot be a Workflow +argument or an Activity result. A :class:`StreamRef` can: it names the owner +and the topic, nothing more, and whoever receives it opens the stream on the +store its own client carries. It carries no cursor, because a position +belongs to a reader, and no store name. +""" + +from __future__ import annotations + +import dataclasses +from dataclasses import dataclass +from typing import Any, Literal + +from temporalio.contrib.streams._topic import ( + DEFAULT_TOPIC, + StreamTopic, + resolve_topic, +) + +__all__ = ["StreamOwnerKind", "StreamRef"] + +StreamOwnerKind = Literal["workflow"] +"""The owner kinds this release opens. Only a Workflow owns a stream. + +:attr:`StreamRef.kind` is a plain string, so a ref that names an owner kind +a later release adds still reaches the ref's own check when it is decoded, +instead of failing the converter's type check. +""" + + +@dataclass(frozen=True) +class StreamRef: + """One stream, named by its owner and its topic. + + ``kind`` says what owns the stream; only ``"workflow"`` exists in this + release. A Workflow ref carries ``workflow_id``. The stream belongs to + the Workflow's run chain, and a read on it follows the chain across + Continue-as-New and ends only when the chain ends. ``topic`` is the topic + a handle opened from the ref addresses when a call names none. + + ``chain_run_id`` picks which chain: the one that holds that run. A + Workflow id can start a new chain once the old one closed, and without + ``chain_run_id`` a ref names the newest chain. It never stops a read at + that run's end. + + The default data converter carries it as JSON, so it can be a Workflow + argument or an Activity result. + """ + + kind: str + workflow_id: str + topic: str = DEFAULT_TOPIC + chain_run_id: str | None = None + + def __post_init__(self) -> None: + """Refuse a ref this release cannot open. + + Raises: + ValueError: ``kind`` is not ``"workflow"``, or ``workflow_id`` or + ``topic`` is empty. + """ + if self.kind != "workflow": + raise ValueError(f"unknown StreamRef kind {self.kind!r}") + if not self.workflow_id: + raise ValueError("a Workflow StreamRef needs a workflow_id") + if not self.topic: + raise ValueError("a StreamRef needs a topic name") + + @classmethod + def for_workflow( + cls, + workflow_id: str, + *, + topic: str | StreamTopic[Any] | None = None, + chain_run_id: str | None = None, + ) -> StreamRef: + """A ref to ``topic`` of ``workflow_id``'s stream. + + Without ``topic`` it names + :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. Without + ``chain_run_id`` it names the Workflow id's newest chain, and with + one the chain that holds that run. + """ + name, _ = resolve_topic(topic) + return cls("workflow", workflow_id, topic=name, chain_run_id=chain_run_id) + + def with_topic(self, topic: str | StreamTopic[Any] | None) -> StreamRef: + """The same owner, naming ``topic`` instead. + + ``None`` names :data:`temporalio.contrib.streams.DEFAULT_TOPIC`. + """ + name, _ = resolve_topic(topic) + return dataclasses.replace(self, topic=name) diff --git a/temporalio/contrib/streams/_topic.py b/temporalio/contrib/streams/_topic.py new file mode 100644 index 0000000000..0f2ec562a5 --- /dev/null +++ b/temporalio/contrib/streams/_topic.py @@ -0,0 +1,108 @@ +"""Typed topic definitions. + +A topic is defined once, at module level, with the type its records decode +to, and the Workflow, its Activities and the client code all refer to that +one definition, the way they share a Signal or an Activity. A plain string +names a topic too, for a name decided at runtime; then the decode hint +travels as ``result_type=`` on each read. + +On the wire a topic is a string, and +:attr:`temporalio.contrib.streams.StreamRecord.topic` is that string. +""" + +from __future__ import annotations + +import unicodedata +from dataclasses import dataclass +from typing import Any, Generic, TypeVar, overload + +__all__ = ["DEFAULT_TOPIC", "StreamTopic", "resolve_topic", "topic"] + +T = TypeVar("T") + +DEFAULT_TOPIC = "output" +"""The topic a call addresses when it names none. + +It is an ordinary name, so naming it explicitly addresses the same topic. +""" + + +@dataclass(frozen=True) +class StreamTopic(Generic[T]): + """A topic of a stream, with the type its records decode to. + + Made with :func:`topic`. Pass it wherever a call takes a topic, and the + record and value types follow from it. + """ + + name: str + result_type: type[T] | None = None + """The type records decode to, or ``None`` for the converter's default.""" + + +@overload +def topic(name: str, result_type: type[T]) -> StreamTopic[T]: ... + + +@overload +def topic(name: str, result_type: None = None) -> StreamTopic[Any]: ... + + +def topic(name: str, result_type: type | None = None) -> StreamTopic[Any]: + """Define a topic. + + Args: + name: The topic's name, as it appears on every record. + result_type: The type records on this topic decode to. Without one, + the payload converter's default applies. + + Raises: + ValueError: ``name`` is empty, longer than 256 UTF-8 bytes, or holds + a control character. + """ + _check_name(name) + return StreamTopic(name, result_type) + + +def resolve_topic( + topic: str | StreamTopic[Any] | None = None, result_type: type | None = None +) -> tuple[str, type | None]: + """The name and decode hint a call means, from either form of topic. + + ``None`` means :data:`DEFAULT_TOPIC`. A handle calls this once at the + top of each call that takes a topic, so a definition and a string reach + the store the same way. + + Raises: + ValueError: A definition was given together with ``result_type``, + which would name two types for one topic, or the name is empty, + longer than 256 UTF-8 bytes, or holds a control character. + """ + if isinstance(topic, StreamTopic): + if result_type is not None: + raise ValueError( + f"topic {topic.name!r} already carries its type; do not pass " + "result_type= with a definition" + ) + name, result_type = topic.name, topic.result_type + elif topic is None: + name = DEFAULT_TOPIC + else: + name = topic + _check_name(name) + return name, result_type + + +_MAX_NAME_BYTES = 256 + + +def _check_name(name: str) -> None: + # The same limit in every SDK: the name goes into store keys, cursor + # hashes and Core's manifest budget, and a store may join names with a + # control character. + if not name: + raise ValueError("topic name must not be empty") + if len(name.encode()) > _MAX_NAME_BYTES: + raise ValueError(f"topic name must be at most {_MAX_NAME_BYTES} UTF-8 bytes") + if any(unicodedata.category(c) == "Cc" for c in name): + raise ValueError(f"topic name {name!r} must not hold a control character") diff --git a/tests/contrib/streams/test_public_surface.py b/tests/contrib/streams/test_public_surface.py index a3647d4eba..1161287e40 100644 --- a/tests/contrib/streams/test_public_surface.py +++ b/tests/contrib/streams/test_public_surface.py @@ -12,6 +12,7 @@ "decode_body", "encode_bodies", "error_from_failure", + "resolve_topic", } diff --git a/tests/contrib/streams/test_topics_refs.py b/tests/contrib/streams/test_topics_refs.py new file mode 100644 index 0000000000..2f51d457b5 --- /dev/null +++ b/tests/contrib/streams/test_topics_refs.py @@ -0,0 +1,100 @@ +"""Typed topics and the stream reference that crosses process boundaries.""" + +from __future__ import annotations + +import pytest + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.streams import ( + DEFAULT_TOPIC, + StreamRef, + StreamTopic, + topic, +) +from temporalio.contrib.streams._topic import ( + resolve_topic, +) +from temporalio.converter import DataConverter + +OUT = topic("out", dict) + + +def test_a_definition_carries_its_name_and_type(): + assert OUT == StreamTopic("out", dict) + assert resolve_topic(OUT) == ("out", dict) + assert resolve_topic("free", int) == ("free", int) + assert resolve_topic(None) == (DEFAULT_TOPIC, None) + + +def test_topic_mistakes_are_value_errors(): + with pytest.raises(ValueError): + topic("") + with pytest.raises(ValueError): + resolve_topic("") + with pytest.raises(ValueError, match="already carries its type"): + resolve_topic(OUT, dict) + + +def test_a_ref_names_a_workflow_stream(): + ref = StreamRef.for_workflow("wf", topic=OUT) + assert ref == StreamRef("workflow", "wf", topic="out") + assert StreamRef.for_workflow("wf").topic == DEFAULT_TOPIC + assert ref.with_topic(None).topic == DEFAULT_TOPIC + assert ref.with_topic("other").workflow_id == "wf" + + +def test_a_ref_round_trips_through_the_default_converter(): + ref = StreamRef.for_workflow("wf", topic=OUT, chain_run_id="run") + converter = DataConverter.default.payload_converter + payload = converter.to_payloads([ref])[0] + assert converter.from_payloads([payload], [StreamRef])[0] == ref + + +def test_a_ref_with_the_old_run_id_decodes_as_the_newest_chain(): + # The old field also ended a read at that run, which no read does now, + # so it is dropped rather than read as chain_run_id. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "workflow", "workflow_id": "wf", "run_id": "r", "topic": "out"}', + ) + ref = converter.from_payloads([payload], [StreamRef])[0] + assert ref == StreamRef.for_workflow("wf", topic=OUT) + + +def test_a_ref_refuses_what_it_cannot_name(): + with pytest.raises(ValueError, match="unknown"): + StreamRef("nexus", "x") + with pytest.raises(ValueError): + StreamRef("workflow", "") + with pytest.raises(ValueError): + StreamRef("workflow", "wf", topic="") + + +def test_a_ref_from_a_later_release_decodes_to_the_kind_check(): + # The kind is a plain string, so the converter builds the ref and the ref + # itself refuses the kind, rather than the converter failing a type check. + converter = DataConverter.default.payload_converter + payload = Payload( + metadata={"encoding": b"json/plain"}, + data=b'{"kind": "activity", "workflow_id": "wf", "topic": "out"}', + ) + with pytest.raises(ValueError, match="'activity'"): + converter.from_payloads([payload], [StreamRef]) + + +@pytest.mark.parametrize( + "name", + ["x" * 257, "é" * 129, "a\x1fb", "line\nbreak", "\x00"], +) +def test_a_topic_name_that_is_too_long_or_has_a_control_character_is_refused(name: str): + # Every store puts the name in keys, cursors and Core's manifest budget. + with pytest.raises(ValueError): + topic(name) + with pytest.raises(ValueError): + resolve_topic(name) + + +def test_a_topic_name_of_256_bytes_is_accepted(): + assert topic("x" * 256).name == "x" * 256 + assert resolve_topic("é" * 128)[0] == "é" * 128