feat: add TS.READ command support to the timeseries module (#4170)

* feat: add TS.READ command support to the timeseries module

* Fix linters after conflict resolution

Author: petyaslavova

redis/commands/timeseries/__init__.py

@@ -24,6 +24,7 @@
     NREVRANGE_CMD,
     QUERYINDEX_CMD,
     RANGE_CMD,
+    READ_CMD,
     REVRANGE_CMD,
     TimeSeriesCommands,
 )
@@ -59,6 +60,7 @@ def __init__(self, client=None, **kwargs):
             CREATE_CMD: bool_ok,
             CREATERULE_CMD: bool_ok,
             DELETERULE_CMD: bool_ok,
+            READ_CMD: parse_range_unified,
             NRANGE_CMD: parse_n_range,
             NREVRANGE_CMD: parse_n_range,
         }
@@ -83,6 +85,7 @@ def __init__(self, client=None, **kwargs):
             MRANGE_CMD: parse_m_range_unified,
             MREVRANGE_CMD: parse_m_range_unified,
             RANGE_CMD: parse_range_unified,
+            READ_CMD: parse_range_unified,
             REVRANGE_CMD: parse_range_unified,
         }
         _RESP3_UNIFIED_MODULE_CALLBACKS = {

redis/commands/timeseries/commands.py

@@ -33,6 +33,7 @@
 NREVRANGE_CMD = "TS.NREVRANGE"
 QUERYINDEX_CMD = "TS.QUERYINDEX"
 RANGE_CMD = "TS.RANGE"
+READ_CMD = "TS.READ"
 REVRANGE_CMD = "TS.REVRANGE"
 
 
@@ -996,6 +997,93 @@ def revrange(
         )
         return self.execute_command(REVRANGE_CMD, *params, keys=[key])
 
+    @overload
+    def read(
+        self: SyncClientProtocol,
+        key: KeyT,
+        timestamp: int | str,
+        block_milliseconds: int | None = None,
+        block_min_count: int | None = None,
+        max_count: int | None = None,
+    ) -> TimeSeriesRangeResponse: ...
+
+    @overload
+    def read(
+        self: AsyncClientProtocol,
+        key: KeyT,
+        timestamp: int | str,
+        block_milliseconds: int | None = None,
+        block_min_count: int | None = None,
+        max_count: int | None = None,
+    ) -> Awaitable[TimeSeriesRangeResponse]: ...
+
+    def read(
+        self,
+        key: KeyT,
+        timestamp: int | str,
+        block_milliseconds: int | None = None,
+        block_min_count: int | None = None,
+        max_count: int | None = None,
+    ) -> TimeSeriesRangeResponse | Awaitable[TimeSeriesRangeResponse]:
+        """
+        Read a batch of samples with timestamps at or after `timestamp`, in
+        ascending timestamp order.
+
+        Without blocking, returns immediately with whatever qualifies (possibly an
+        empty list). With the `block_milliseconds` group, waits until at least
+        `block_min_count` qualifying samples exist or until the timeout elapses.
+        This allows consuming historical and newly-appended samples continuously,
+        in batches, without polling `TS.RANGE`.
+
+        For more information see https://redis.io/commands/ts.read/
+
+        Args:
+            key:
+                Key name for the time series (a regular series or a compaction
+                destination).
+            timestamp:
+                Inclusive cursor. Samples with `timestamp >= timestamp` qualify.
+                A non-negative integer (Unix milliseconds, `0` reads from the
+                beginning) or one of the sentinels `-` (earliest), `+` (latest
+                existing sample, inclusive) or `$` (only samples added after the
+                command is received). Sentinels are sent to the server as-is and
+                resolved server-side; `$` is only meaningful together with
+                `block_milliseconds`, since nothing can qualify at execution time
+                without blocking.
+            block_milliseconds:
+                Opt into blocking. Maximum time to wait, in whole milliseconds;
+                a non-negative integer where `0` means wait indefinitely. When
+                `None` (default) the command does not block.
+            block_min_count:
+                The unblock threshold: the call returns once this many samples
+                qualify. A positive integer, defaulting to `1` when blocking is
+                requested. Only used when `block_milliseconds` is set; the value is
+                always emitted on the wire inside the BLOCK group.
+            max_count:
+                Reply cap, a positive integer. When more samples qualify than
+                `max_count`, the oldest `max_count` are returned so callers can page
+                forward. `None` (default) means unlimited.
+
+        Returns:
+            A list of `[timestamp, value]` samples in ascending timestamp order.
+            An empty list is a successful reply (returned when nothing qualifies,
+            or when a blocking call times out with nothing available).
+
+        .. warning::
+            A blocking call keeps the connection parked for up to
+            `block_milliseconds`. The client's `socket_timeout` still applies: with
+            the default (5 seconds) a longer block raises `TimeoutError` before the
+            server replies. When using `block_milliseconds`, configure the client
+            with a `socket_timeout` larger than the block window (or `None`), as with
+            other blocking commands. This command must not be retried automatically
+            after an empty or partial reply.
+        """
+        params: list[EncodableT] = [key, timestamp]
+        self._append_block(params, block_milliseconds, block_min_count)
+        self._append_max_count(params, max_count)
+
+        return self.execute_command(READ_CMD, *params, keys=[key])
+
     def __n_range_params(
         self,
         keys: List[KeyT],
@@ -1785,6 +1873,35 @@ def _append_count(params: list[EncodableT], count: int | None):
         if count is not None:
             params.extend(["COUNT", count])
 
+    @staticmethod
+    def _append_block(
+        params: list[EncodableT],
+        block_milliseconds: int | None,
+        block_min_count: int | None,
+    ):
+        """Append the BLOCK group to params.
+
+        The BLOCK group is all-or-nothing: when blocking is requested
+        (`block_milliseconds` is set), both `milliseconds` and `min_count` are
+        always emitted, with `min_count` defaulting to 1. There is no standalone
+        MIN_COUNT keyword in this command.
+        """
+        if block_milliseconds is None:
+            if block_min_count is not None:
+                raise DataError(
+                    "block_min_count requires block_milliseconds to be set; the "
+                    "BLOCK group is all-or-nothing."
+                )
+            return
+        min_count = 1 if block_min_count is None else block_min_count
+        params.extend(["BLOCK", block_milliseconds, min_count])
+
+    @staticmethod
+    def _append_max_count(params: list[EncodableT], max_count: int | None):
+        """Append MAX_COUNT property to params."""
+        if max_count is not None:
+            params.extend(["MAX_COUNT", max_count])
+
     @staticmethod
     def _append_timestamp(params: list[EncodableT], timestamp: int | None):
         """Append TIMESTAMP property to params."""

tests/test_asyncio/test_timeseries.py

@@ -365,6 +365,94 @@ async def test_rev_range(decoded_r: redis.Redis):
     )
 
 
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+async def test_read(decoded_r: redis.Redis):
+    await decoded_r.ts().create(1)
+    await decoded_r.ts().add(1, 100, 1.0)
+    await decoded_r.ts().add(1, 200, 2.0)
+    await decoded_r.ts().add(1, 300, 3.0)
+
+    # Read everything at or after the cursor. TS.READ always returns the same
+    # unified sample shape (list of [timestamp, value]) regardless of protocol.
+    assert await decoded_r.ts().read(1, 0) == [[100, 1.0], [200, 2.0], [300, 3.0]]
+
+    # The cursor is inclusive.
+    assert await decoded_r.ts().read(1, 200) == [[200, 2.0], [300, 3.0]]
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+async def test_read_max_count(decoded_r: redis.Redis):
+    await decoded_r.ts().create(1)
+    await decoded_r.ts().add(1, 100, 1.0)
+    await decoded_r.ts().add(1, 200, 2.0)
+    await decoded_r.ts().add(1, 300, 3.0)
+
+    # Bounded paging: read the oldest max_count, then page from last_ts + 1.
+    assert await decoded_r.ts().read(1, "-", max_count=2) == [[100, 1.0], [200, 2.0]]
+    assert await decoded_r.ts().read(1, 201, max_count=2) == [[300, 3.0]]
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+async def test_read_sentinels(decoded_r: redis.Redis):
+    await decoded_r.ts().create(1)
+    await decoded_r.ts().add(1, 100, 1.0)
+    await decoded_r.ts().add(1, 200, 2.0)
+    await decoded_r.ts().add(1, 300, 3.0)
+
+    # `+` resolves to the latest sample, inclusive; returned even without BLOCK.
+    assert await decoded_r.ts().read(1, "+") == [[300, 3.0]]
+
+    # `-` reads from the earliest sample.
+    assert len(await decoded_r.ts().read(1, "-")) == 3
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+async def test_read_empty(decoded_r: redis.Redis):
+    await decoded_r.ts().create(1)
+    await decoded_r.ts().add(1, 100, 1.0)
+
+    # A cursor past the newest sample yields an empty (successful) reply.
+    assert [] == await decoded_r.ts().read(1, 301)
+    # A missing key is also an empty reply, not an error.
+    assert [] == await decoded_r.ts().read("missing", 0)
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+async def test_read_block(decoded_r: redis.Redis):
+    await decoded_r.ts().create(1)
+    await decoded_r.ts().add(1, 100, 1.0)
+    await decoded_r.ts().add(1, 200, 2.0)
+    await decoded_r.ts().add(1, 300, 3.0)
+
+    # min_count is already met, so the blocking call returns immediately.
+    res = await decoded_r.ts().read(1, 0, block_milliseconds=1000, block_min_count=1)
+    assert len(res) == 3
+
+    # min_count cannot be reached; after the timeout the available samples flush.
+    res = await decoded_r.ts().read(1, 101, block_milliseconds=100, block_min_count=10)
+    assert res == [[200, 2.0], [300, 3.0]]
+
+    # A blocking timeout with nothing available is a successful empty reply.
+    assert [] == await decoded_r.ts().read(
+        1, 301, block_milliseconds=100, block_min_count=1
+    )
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+async def test_read_block_min_count_requires_milliseconds(decoded_r: redis.Redis):
+    # BLOCK is all-or-nothing: min_count without milliseconds is invalid usage.
+    with pytest.raises(
+        redis.DataError, match="block_min_count requires block_milliseconds"
+    ):
+        await decoded_r.ts().read(1, 0, block_min_count=5)
+
+
 @pytest.mark.onlynoncluster
 @pytest.mark.redismod
 async def test_multi_range(decoded_r: redis.Redis):

tests/test_timeseries.py

@@ -551,6 +551,92 @@ def test_revrange_empty(client: redis.Redis):
     assert_resp_response(client, res, resp2_expected, resp3_expected)
 
 
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+def test_read(client):
+    client.ts().create(1)
+    client.ts().add(1, 100, 1.0)
+    client.ts().add(1, 200, 2.0)
+    client.ts().add(1, 300, 3.0)
+
+    # Read everything at or after the cursor. TS.READ always returns the same
+    # unified sample shape (list of [timestamp, value]) regardless of protocol.
+    assert client.ts().read(1, 0) == [[100, 1.0], [200, 2.0], [300, 3.0]]
+
+    # The cursor is inclusive.
+    assert client.ts().read(1, 200) == [[200, 2.0], [300, 3.0]]
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+def test_read_max_count(client):
+    client.ts().create(1)
+    client.ts().add(1, 100, 1.0)
+    client.ts().add(1, 200, 2.0)
+    client.ts().add(1, 300, 3.0)
+
+    # Bounded paging: read the oldest max_count, then page from last_ts + 1.
+    assert client.ts().read(1, "-", max_count=2) == [[100, 1.0], [200, 2.0]]
+    assert client.ts().read(1, 201, max_count=2) == [[300, 3.0]]
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+def test_read_sentinels(client):
+    client.ts().create(1)
+    client.ts().add(1, 100, 1.0)
+    client.ts().add(1, 200, 2.0)
+    client.ts().add(1, 300, 3.0)
+
+    # `+` resolves to the latest sample, inclusive; returned even without BLOCK.
+    assert client.ts().read(1, "+") == [[300, 3.0]]
+
+    # `-` reads from the earliest sample.
+    assert len(client.ts().read(1, "-")) == 3
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+def test_read_empty(client):
+    client.ts().create(1)
+    client.ts().add(1, 100, 1.0)
+
+    # A cursor past the newest sample yields an empty (successful) reply.
+    assert [] == client.ts().read(1, 301)
+    # A missing key is also an empty reply, not an error.
+    assert [] == client.ts().read("missing", 0)
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+def test_read_block(client):
+    client.ts().create(1)
+    client.ts().add(1, 100, 1.0)
+    client.ts().add(1, 200, 2.0)
+    client.ts().add(1, 300, 3.0)
+
+    # min_count is already met, so the blocking call returns immediately.
+    res = client.ts().read(1, 0, block_milliseconds=1000, block_min_count=1)
+    assert 3 == len(res)
+
+    # min_count cannot be reached; after the timeout the available samples flush.
+    res = client.ts().read(1, 101, block_milliseconds=100, block_min_count=10)
+    assert res == [[200, 2.0], [300, 3.0]]
+
+    # A blocking timeout with nothing available is a successful empty reply.
+    assert [] == client.ts().read(1, 301, block_milliseconds=100, block_min_count=1)
+
+
+@pytest.mark.redismod
+@skip_if_server_version_lt("8.9.0")
+def test_read_block_min_count_requires_milliseconds(client):
+    # BLOCK is all-or-nothing: min_count without milliseconds is invalid usage.
+    with pytest.raises(
+        redis.exceptions.DataError, match="block_min_count requires block_milliseconds"
+    ):
+        client.ts().read(1, 0, block_min_count=5)
+
+
 @pytest.mark.onlynoncluster
 @pytest.mark.redismod
 def test_mrange(client):