subscribe-sdk: Add the Python SDK package

misc/subscribe-sdk/python/.gitignore

@@ -0,0 +1,8 @@
+# Python caches, builds, and virtualenvs.
+__pycache__/
+*.pyc
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+*.egg-info/
+.venv/

misc/subscribe-sdk/python/README.md

@@ -0,0 +1,66 @@
+# materialize-subscribe
+
+A correct-by-construction Python client for consuming Materialize
+[`SUBSCRIBE`](https://materialize.com/docs/sql/subscribe/).
+
+```python
+from materialize_subscribe import Subscribe, SubscribeClient
+
+with SubscribeClient.connect("postgres://materialize@localhost:6875") as client:
+    stream = client.subscribe(
+        Subscribe.object("winning_bids").envelope_upsert(["id"])
+    )
+    for batch in stream:
+        for change in batch.updates:
+            ...  # apply `change` to your sink
+        # Persist `batch.resume_token.encode()` atomically with the applied
+        # changes for exactly-once state; hand it back to `client.resume`.
+```
+
+* The unit of consumption is a `ConsistentBatch`, never a bare row.
+* Resuming takes an opaque `ResumeToken`; the `AS OF frontier - 1` arithmetic
+  lives inside it.
+* Failures surface as a small, typed `SubscribeError` hierarchy.
+* For a lower-level stream of decoded changes and progress markers, use
+  `subscribe_raw`. The batcher is built on it.
+
+## Cohorts
+
+To stay consistent across several views at once, a `Cohort` releases every
+member only up to their shared minimum frontier, so each moment is a genuine
+cross-view snapshot.
+
+```python
+from materialize_subscribe import Cohort, Subscribe
+
+with Cohort.connect(
+    "postgres://materialize@localhost:6875",
+    [
+        ("orders", Subscribe.object("orders")),
+        ("inventory", Subscribe.object("inventory")),
+    ],
+) as cohort:
+    for moment in cohort:
+        for view in moment.views:
+            ...  # view.name, view.updates at the joint frontier
+        # moment.resume_token checkpoints the whole cohort at once.
+```
+
+The protocol core has no runtime dependencies. Install the `client` extra to
+connect:
+
+```
+pip install 'materialize-subscribe[client]'
+```
+
+See the [top-level README](../README.md) for the protocol this encodes and the
+guarantees it provides.
+
+## Development
+
+```
+pip install -e '.[dev,client]'
+pytest
+mypy materialize_subscribe
+ruff check .
+```

misc/subscribe-sdk/python/materialize_subscribe/__init__.py

@@ -0,0 +1,118 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License in the LICENSE file at the
+# root of this repository, or online at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A correct-by-construction client for consuming Materialize ``SUBSCRIBE``.
+
+Consuming ``SUBSCRIBE`` durably requires a specific protocol: buffer until a
+progress message closes a timestamp, apply the closed batch and persist its
+frontier atomically, and resume with ``SNAPSHOT = false AS OF frontier - 1``.
+Every step has a failure mode that is invisible in testing. This package
+encodes the protocol so callers cannot get it wrong:
+
+* The unit of consumption is a :class:`ConsistentBatch`, never a bare row.
+* Resuming takes an opaque :class:`ResumeToken`; the ``- 1`` boundary
+  arithmetic lives inside it.
+* Failures surface as a small, typed :class:`SubscribeError` hierarchy.
+
+The client is three layers, each usable on its own: :class:`RawStream` (via
+:meth:`SubscribeClient.subscribe_raw`) is the decoded timestamped substrate; the
+consistency engine buffers and releases below a frontier; a
+:class:`ConsistentBatch` (one view) or :class:`CohortMoment` (a :class:`Cohort`
+of N views) is the closed unit. A single view is the cohort of one.
+
+Example::
+
+    from materialize_subscribe import Subscribe, SubscribeClient
+
+    with SubscribeClient.connect("postgres://materialize@localhost:6875") as client:
+        stream = client.subscribe(
+            Subscribe.object("winning_bids").envelope_upsert(["id"])
+        )
+        for batch in stream:
+            for change in batch.updates:
+                ...  # apply `change` to your sink
+            # Persist `batch.resume_token.encode()` atomically with the applied
+            # changes for exactly-once state; hand it back to `client.resume`.
+"""
+
+from __future__ import annotations
+
+from .batch import Batcher, ConsistentBatch
+from .client import BatchStream, RawStream, SubscribeClient
+from .cohort import Cohort, CohortMoment, ViewChanges
+from .envelope import (
+    Change,
+    Data,
+    Decoder,
+    Delete,
+    Diff,
+    DiffEnvelope,
+    Envelope,
+    KeyViolation,
+    Progress,
+    StreamMessage,
+    Upsert,
+    UpsertEnvelope,
+)
+from .errors import (
+    BufferOverflow,
+    CohortLagExceeded,
+    CompactionHorizon,
+    DependencyDropped,
+    FatalError,
+    InvalidToken,
+    ProtocolError,
+    SchemaMismatch,
+    SubscribeError,
+    TransientError,
+)
+from .statement import Subscribe
+from .token import CohortToken, ResumeToken
+
+__all__ = [
+    "Batcher",
+    "BatchStream",
+    "BufferOverflow",
+    "Change",
+    "Cohort",
+    "CohortLagExceeded",
+    "CohortMoment",
+    "CohortToken",
+    "CompactionHorizon",
+    "ConsistentBatch",
+    "Data",
+    "Decoder",
+    "Delete",
+    "DependencyDropped",
+    "Diff",
+    "DiffEnvelope",
+    "Envelope",
+    "FatalError",
+    "InvalidToken",
+    "KeyViolation",
+    "Progress",
+    "ProtocolError",
+    "RawStream",
+    "ResumeToken",
+    "SchemaMismatch",
+    "StreamMessage",
+    "Subscribe",
+    "SubscribeClient",
+    "SubscribeError",
+    "TransientError",
+    "Upsert",
+    "UpsertEnvelope",
+    "ViewChanges",
+]

misc/subscribe-sdk/python/materialize_subscribe/batch.py

@@ -0,0 +1,140 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License in the LICENSE file at the
+# root of this repository, or online at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""The progress-driven batcher: the state machine that turns a stream of
+changes and progress markers into *consistent, closed* batches.
+
+This is the part every hand-rolled subscribe consumer gets wrong. The rule it
+enforces:
+
+* Buffer changes as they arrive.
+* When a progress marker advances the frontier to ``F``, every change with a
+  timestamp strictly below ``F`` is now final. Emit exactly those as one batch,
+  tagged with frontier ``F`` and a resume token for ``F``.
+* Never emit a change before its timestamp is closed, and never emit the same
+  timestamp twice.
+
+A consumer that applies each emitted batch and persists its token atomically
+gets exactly-once *state*: after any crash, resuming from the last token neither
+drops nor duplicates data.
+
+A single view is the one-member case of the cohort engine (see
+:mod:`materialize_subscribe.cohort`): the release frontier is the view's own
+progress frontier. Both share the :class:`ReleaseBuffer` buffer-and-consolidate
+core.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import List, Optional
+
+from .engine import ReleaseBuffer
+from .envelope import Change, Data, Envelope, Progress, StreamMessage
+from .errors import ProtocolError
+from .token import ResumeToken
+
+
+@dataclass(frozen=True)
+class ConsistentBatch:
+    """A batch of changes for a contiguous, now-closed range of timestamps.
+
+    Every change in ``updates`` has a timestamp strictly below ``frontier``, and
+    no future batch will contain a timestamp below ``frontier``. ``resume_token``
+    checkpoints exactly this position. ``updates`` is consolidated to its net
+    effect within the batch, ordered by first appearance.
+    """
+
+    updates: List[Change]
+    frontier: int
+    resume_token: ResumeToken
+    is_snapshot: bool
+
+    def is_empty(self) -> bool:
+        """Whether the batch carries no changes.
+
+        Empty batches still advance the frontier and carry a fresh token, which
+        lets a consumer keep its checkpoint moving during idle periods so it
+        does not age out of the source's retained history.
+        """
+        return not self.updates
+
+
+class Batcher:
+    """Accumulates stream messages and emits :class:`ConsistentBatch` values as
+    timestamps close.
+
+    Feed every decoded message to :meth:`push`. It returns a batch when a
+    progress marker closes one or more timestamps, and ``None`` otherwise.
+    """
+
+    def __init__(
+        self, fingerprint: str, envelope: Envelope, with_snapshot: bool
+    ) -> None:
+        self._fingerprint = fingerprint
+        self._with_snapshot = with_snapshot
+        self._buffer = ReleaseBuffer(envelope)
+        self._last_frontier: Optional[int] = None
+        self._first_frontier: Optional[int] = None
+        self._snapshot_emitted = False
+
+    def push(self, message: StreamMessage) -> Optional[ConsistentBatch]:
+        """Feeds one decoded message. Returns a batch when the frontier
+        advances."""
+        if isinstance(message, Data):
+            self._buffer.push_data(message.timestamp, message.change)
+            return None
+
+        if isinstance(message, Progress):
+            return self._advance(message.frontier)
+
+        raise ProtocolError(f"unexpected stream message: {message!r}")
+
+    def _advance(self, frontier: int) -> Optional[ConsistentBatch]:
+        if frontier == self._last_frontier:
+            # No advance: nothing new is closed. A regression, by contrast, is
+            # caught by `observe_progress` below.
+            return None
+        self._buffer.observe_progress(frontier)
+
+        if self._first_frontier is None:
+            self._first_frontier = frontier
+
+        updates = self._buffer.release_below(frontier)
+
+        # The snapshot is the data closed once we advance past the first frontier
+        # (the effective AS OF). Mark the first such batch.
+        #
+        # NOTE: this relies on the server emitting a leading Progress(as_of)
+        # before any snapshot rows (it does, unconditionally, when PROGRESS is
+        # set). That progress sets first_frontier; the snapshot rows sit at
+        # as_of and are closed by the next, higher progress. A transport that
+        # delivered snapshot data before any progress would mistag it.
+        is_snapshot = (
+            self._with_snapshot
+            and not self._snapshot_emitted
+            and frontier != self._first_frontier
+        )
+        if is_snapshot:
+            self._snapshot_emitted = True
+
+        self._last_frontier = frontier
+
+        return ConsistentBatch(
+            updates=updates,
+            frontier=frontier,
+            resume_token=ResumeToken(frontier=frontier, fingerprint=self._fingerprint),
+            is_snapshot=is_snapshot,
+        )