Cover TS.INCRBY and TS.DECRBY

Author: gerzse

redis/commands/timeseries/commands.py

@@ -88,7 +88,7 @@ def create(
         self._append_retention(params, retention_msecs)
         self._append_uncompressed(params, uncompressed)
         self._append_chunk_size(params, chunk_size)
-        self._append_duplicate_policy(params, CREATE_CMD, duplicate_policy)
+        self._append_duplicate_policy(params, duplicate_policy)
         self._append_labels(params, labels)
         self._append_ignore_filters(params, ignore_max_time_diff, ignore_max_val_diff)
 
@@ -98,6 +98,7 @@ def alter(
         self,
         key: KeyT,
         retention_msecs: Optional[int] = None,
+        uncompressed: Optional[bool] = False,
         labels: Optional[Dict[str, str]] = None,
         chunk_size: Optional[int] = None,
         duplicate_policy: Optional[str] = None,
@@ -117,6 +118,8 @@ def alter(
             retention_msecs:
                 Maximum age for samples, compared to the highest reported timestamp in
                 milliseconds. If None or 0 is passed, the series is not trimmed at all.
+            uncompressed:
+                Changes data storage from compressed (default) to uncompressed.
             labels:
                 A dictionary of label-value pairs that represent metadata labels of the
                 key.
@@ -153,8 +156,9 @@ def alter(
         """
         params = [key]
         self._append_retention(params, retention_msecs)
+        self._append_uncompressed(params, uncompressed)
         self._append_chunk_size(params, chunk_size)
-        self._append_duplicate_policy(params, ALTER_CMD, duplicate_policy)
+        self._append_duplicate_policy(params, duplicate_policy)
         self._append_labels(params, labels)
         self._append_ignore_filters(params, ignore_max_time_diff, ignore_max_val_diff)
 
@@ -172,6 +176,7 @@ def add(
         duplicate_policy: Optional[str] = None,
         ignore_max_time_diff: Optional[int] = None,
         ignore_max_val_diff: Optional[Number] = None,
+        on_duplicate: Optional[str] = None,
     ):
         """
         Append (or create and append) a new sample to a time series.
@@ -225,14 +230,18 @@ def add(
                 is lower than this threshold, the new entry is ignored. Only applicable
                 if `duplicate_policy` is set to `last`, and if `ignore_max_time_diff` is
                 also set. Available since RedisTimeSeries version 1.12.0.
+            on_duplicate:
+                Use a specific duplicate policy for the specified timestamp. Overrides
+                the duplicate policy set by `duplicate_policy`.
         """
         params = [key, timestamp, value]
         self._append_retention(params, retention_msecs)
         self._append_uncompressed(params, uncompressed)
         self._append_chunk_size(params, chunk_size)
-        self._append_duplicate_policy(params, ADD_CMD, duplicate_policy)
+        self._append_duplicate_policy(params, duplicate_policy)
         self._append_labels(params, labels)
         self._append_ignore_filters(params, ignore_max_time_diff, ignore_max_val_diff)
+        self._append_on_duplicate(params, on_duplicate)
 
         return self.execute_command(ADD_CMD, *params)
 
@@ -260,6 +269,9 @@ def incrby(
         uncompressed: Optional[bool] = False,
         labels: Optional[Dict[str, str]] = None,
         chunk_size: Optional[int] = None,
+        duplicate_policy: Optional[str] = None,
+        ignore_max_time_diff: Optional[int] = None,
+        ignore_max_val_diff: Optional[Number] = None,
     ):
         """
         Increment (or create an time-series and increment) the latest sample's of a series.
@@ -276,21 +288,52 @@ def incrby(
                 Timestamp of the sample. `*` can be used for automatic timestamp (using
                 the system clock).
             retention_msecs:
-                Maximum age for samples compared to last event time (in milliseconds).
-                If `None` or `0` is passed then  the series is not trimmed at all.
+                Maximum age for samples, compared to the highest reported timestamp in
+                milliseconds. If None or 0 is passed, the series is not trimmed at all.
             uncompressed:
-                Changes data storage from compressed (by default) to uncompressed.
+                Changes data storage from compressed (default) to uncompressed.
             labels:
-                Set of label-value pairs that represent metadata labels of the key.
+                A dictionary of label-value pairs that represent metadata labels of the
+                key.
             chunk_size:
-                Memory size, in bytes, allocated for each data chunk.
+                Memory size, in bytes, allocated for each data chunk. Must be a multiple
+                of 8 in the range [128..1048576].
+            duplicate_policy:
+                Policy for handling multiple samples with identical timestamps. Can be
+                one of:
+                    - 'block': An error will occur for any out of order sample.
+                    - 'first': Ignore the new value.
+                    - 'last': Override with the latest value.
+                    - 'min': Only override if the value is lower than the existing
+                      value.
+                    - 'max': Only override if the value is higher than the existing
+                      value.
+                    - 'sum': If a previous sample exists, add the new sample to it so
+                      that the updated value is equal to (previous + new). If no
+                      previous sample exists, set the updated value equal to the new
+                      value.
+            ignore_max_time_diff:
+                A non-negative integer value, in milliseconds, that sets an ignore
+                threshold for added timestamps. If the difference between the last
+                timestamp and the new timestamp is lower than this threshold, the new
+                entry is ignored. Only applicable if `duplicate_policy` is set to
+                `last`, and if `ignore_max_val_diff` is also set. Available since
+                RedisTimeSeries version 1.12.0.
+            ignore_max_val_diff:
+                A non-negative floating point value, that sets an ignore threshold for
+                added values. If the difference between the last value and the new value
+                is lower than this threshold, the new entry is ignored. Only applicable
+                if `duplicate_policy` is set to `last`, and if `ignore_max_time_diff` is
+                also set. Available since RedisTimeSeries version 1.12.0.
         """
         params = [key, value]
         self._append_timestamp(params, timestamp)
         self._append_retention(params, retention_msecs)
         self._append_uncompressed(params, uncompressed)
         self._append_chunk_size(params, chunk_size)
+        self._append_duplicate_policy(params, duplicate_policy)
         self._append_labels(params, labels)
+        self._append_ignore_filters(params, ignore_max_time_diff, ignore_max_val_diff)
 
         return self.execute_command(INCRBY_CMD, *params)
 
@@ -303,6 +346,9 @@ def decrby(
         uncompressed: Optional[bool] = False,
         labels: Optional[Dict[str, str]] = None,
         chunk_size: Optional[int] = None,
+        duplicate_policy: Optional[str] = None,
+        ignore_max_time_diff: Optional[int] = None,
+        ignore_max_val_diff: Optional[Number] = None,
     ):
         """
         Decrement (or create an time-series and decrement) the latest sample's of a series.
@@ -319,21 +365,52 @@ def decrby(
                 Timestamp of the sample. `*` can be used for automatic timestamp (using
                 the system clock).
             retention_msecs:
-                Maximum age for samples compared to last event time (in milliseconds).
-                If `None` or `0` is passed then the series is not trimmed at all.
+                Maximum age for samples, compared to the highest reported timestamp in
+                milliseconds. If None or 0 is passed, the series is not trimmed at all.
             uncompressed:
-                Changes data storage from compressed (by default) to uncompressed.
+                Changes data storage from compressed (default) to uncompressed.
             labels:
-                Set of label-value pairs that represent metadata labels of the key.
+                A dictionary of label-value pairs that represent metadata labels of the
+                key.
             chunk_size:
-                Memory size, in bytes, allocated for each data chunk.
+                Memory size, in bytes, allocated for each data chunk. Must be a multiple
+                of 8 in the range [128..1048576].
+            duplicate_policy:
+                Policy for handling multiple samples with identical timestamps. Can be
+                one of:
+                    - 'block': An error will occur for any out of order sample.
+                    - 'first': Ignore the new value.
+                    - 'last': Override with the latest value.
+                    - 'min': Only override if the value is lower than the existing
+                      value.
+                    - 'max': Only override if the value is higher than the existing
+                      value.
+                    - 'sum': If a previous sample exists, add the new sample to it so
+                      that the updated value is equal to (previous + new). If no
+                      previous sample exists, set the updated value equal to the new
+                      value.
+            ignore_max_time_diff:
+                A non-negative integer value, in milliseconds, that sets an ignore
+                threshold for added timestamps. If the difference between the last
+                timestamp and the new timestamp is lower than this threshold, the new
+                entry is ignored. Only applicable if `duplicate_policy` is set to
+                `last`, and if `ignore_max_val_diff` is also set. Available since
+                RedisTimeSeries version 1.12.0.
+            ignore_max_val_diff:
+                A non-negative floating point value, that sets an ignore threshold for
+                added values. If the difference between the last value and the new value
+                is lower than this threshold, the new entry is ignored. Only applicable
+                if `duplicate_policy` is set to `last`, and if `ignore_max_time_diff` is
+                also set. Available since RedisTimeSeries version 1.12.0.
         """
         params = [key, value]
         self._append_timestamp(params, timestamp)
         self._append_retention(params, retention_msecs)
         self._append_uncompressed(params, uncompressed)
         self._append_chunk_size(params, chunk_size)
+        self._append_duplicate_policy(params, duplicate_policy)
         self._append_labels(params, labels)
+        self._append_ignore_filters(params, ignore_max_time_diff, ignore_max_val_diff)
 
         return self.execute_command(DECRBY_CMD, *params)
 
@@ -942,17 +1019,16 @@ def _append_chunk_size(params: List[str], chunk_size: Optional[int]):
             params.extend(["CHUNK_SIZE", chunk_size])
 
     @staticmethod
-    def _append_duplicate_policy(
-        params: List[str], command: Optional[str], duplicate_policy: Optional[str]
-    ):
-        """Append DUPLICATE_POLICY property to params on CREATE
-        and ON_DUPLICATE on ADD.
-        """
+    def _append_duplicate_policy(params: List[str], duplicate_policy: Optional[str]):
+        """Append DUPLICATE_POLICY property to params."""
         if duplicate_policy is not None:
-            if command == "TS.ADD":
-                params.extend(["ON_DUPLICATE", duplicate_policy])
-            else:
-                params.extend(["DUPLICATE_POLICY", duplicate_policy])
+            params.extend(["DUPLICATE_POLICY", duplicate_policy])
+
+    @staticmethod
+    def _append_on_duplicate(params: List[str], on_duplicate: Optional[str]):
+        """Append ON_DUPLICATE property to params."""
+        if on_duplicate is not None:
+            params.extend(["ON_DUPLICATE", on_duplicate])
 
     @staticmethod
     def _append_filer_by_ts(params: List[str], ts_list: Optional[List[int]]):

tests/test_timeseries.py

@@ -3,8 +3,8 @@
 from time import sleep
 
 import pytest
-
 import redis
+
 from .conftest import assert_resp_response, is_resp2_connection, skip_ifmodversion_lt
 
 
@@ -103,38 +103,32 @@ def test_add(client):
 
 
 @skip_ifmodversion_lt("1.4.0", "timeseries")
-def test_add_duplicate_policy(client):
+def test_add_on_duplicate(client):
     # Test for duplicate policy BLOCK
     assert 1 == client.ts().add("time-serie-add-ooo-block", 1, 5.0)
     with pytest.raises(Exception):
         client.ts().add("time-serie-add-ooo-block", 1, 5.0, duplicate_policy="block")
 
     # Test for duplicate policy LAST
     assert 1 == client.ts().add("time-serie-add-ooo-last", 1, 5.0)
-    assert 1 == client.ts().add(
-        "time-serie-add-ooo-last", 1, 10.0, duplicate_policy="last"
-    )
+    assert 1 == client.ts().add("time-serie-add-ooo-last", 1, 10.0, on_duplicate="last")
     assert 10.0 == client.ts().get("time-serie-add-ooo-last")[1]
 
     # Test for duplicate policy FIRST
     assert 1 == client.ts().add("time-serie-add-ooo-first", 1, 5.0)
     assert 1 == client.ts().add(
-        "time-serie-add-ooo-first", 1, 10.0, duplicate_policy="first"
+        "time-serie-add-ooo-first", 1, 10.0, on_duplicate="first"
     )
     assert 5.0 == client.ts().get("time-serie-add-ooo-first")[1]
 
     # Test for duplicate policy MAX
     assert 1 == client.ts().add("time-serie-add-ooo-max", 1, 5.0)
-    assert 1 == client.ts().add(
-        "time-serie-add-ooo-max", 1, 10.0, duplicate_policy="max"
-    )
+    assert 1 == client.ts().add("time-serie-add-ooo-max", 1, 10.0, on_duplicate="max")
     assert 10.0 == client.ts().get("time-serie-add-ooo-max")[1]
 
     # Test for duplicate policy MIN
     assert 1 == client.ts().add("time-serie-add-ooo-min", 1, 5.0)
-    assert 1 == client.ts().add(
-        "time-serie-add-ooo-min", 1, 10.0, duplicate_policy="min"
-    )
+    assert 1 == client.ts().add("time-serie-add-ooo-min", 1, 10.0, on_duplicate="min")
     assert 5.0 == client.ts().get("time-serie-add-ooo-min")[1]
 
 
@@ -982,14 +976,11 @@ def test_create_with_insertion_filters(client):
     )
     assert 1000 == client.ts().add("time-series-1", 1000, 1.0)
     assert 1010 == client.ts().add("time-series-1", 1010, 11.0)
-    # Within 5 ms of the last timestamp and value diff less than 10.0
     assert 1010 == client.ts().add("time-series-1", 1013, 10.0)
-    # Value difference less than 10.0, but timestamp diff larger than 5 ms
     assert 1020 == client.ts().add("time-series-1", 1020, 11.5)
-    # Timestamp diff less than 5 ms, but value diff larger than 10.0
     assert 1021 == client.ts().add("time-series-1", 1021, 22.0)
 
-    data_points = client.ts().range("time-series-1", '-', '+')
+    data_points = client.ts().range("time-series-1", "-", "+")
     expected_points = [(1000, 1.0), (1010, 11.0), (1020, 11.5), (1021, 22.0)]
     assert expected_points == data_points
 
@@ -1003,11 +994,10 @@ def test_create_with_insertion_filters_other_duplicate_policy(client):
     )
     assert 1000 == client.ts().add("time-series-1", 1000, 1.0)
     assert 1010 == client.ts().add("time-series-1", 1010, 11.0)
-    # Within 5 ms of the last timestamp and value diff less than 10.0.
     # Still accepted because the duplicate_policy is not `last`.
     assert 1013 == client.ts().add("time-series-1", 1013, 10.0)
 
-    data_points = client.ts().range("time-series-1", '-', '+')
+    data_points = client.ts().range("time-series-1", "-", "+")
     expected_points = [(1000, 1.0), (1010, 11.0), (1013, 10)]
     assert expected_points == data_points
 
@@ -1022,13 +1012,12 @@ def test_alter_with_insertion_filters(client):
         "time-series-1",
         duplicate_policy="last",
         ignore_max_time_diff=5,
-        ignore_max_val_diff=10.0
+        ignore_max_val_diff=10.0,
     )
 
-    # Within 5 ms of the last timestamp and value diff less than 10.0.
     assert 1013 == client.ts().add("time-series-1", 1015, 11.5)
 
-    data_points = client.ts().range("time-series-1", '-', '+')
+    data_points = client.ts().range("time-series-1", "-", "+")
     expected_points = [(1000, 1.0), (1010, 11.0), (1013, 10.0)]
     assert expected_points == data_points
 
@@ -1044,9 +1033,56 @@ def test_add_with_insertion_filters(client):
         ignore_max_val_diff=10.0,
     )
 
-    # Within 5 ms of the last timestamp and value diff less than 10.0.
     assert 1000 == client.ts().add("time-series-1", 1004, 3.0)
 
-    data_points = client.ts().range("time-series-1", '-', '+')
+    data_points = client.ts().range("time-series-1", "-", "+")
+    expected_points = [(1000, 1.0)]
+    assert expected_points == data_points
+
+
+@skip_ifmodversion_lt("1.12.0", "timeseries")
+def test_incrby_with_insertion_filters(client):
+    assert 1000 == client.ts().incrby(
+        "time-series-1",
+        1.0,
+        timestamp=1000,
+        duplicate_policy="last",
+        ignore_max_time_diff=5,
+        ignore_max_val_diff=10.0,
+    )
+
+    assert 1000 == client.ts().incrby("time-series-1", 3.0, timestamp=1000)
+
+    data_points = client.ts().range("time-series-1", "-", "+")
     expected_points = [(1000, 1.0)]
     assert expected_points == data_points
+
+    assert 1000 == client.ts().incrby("time-series-1", 10.1, timestamp=1000)
+
+    data_points = client.ts().range("time-series-1", "-", "+")
+    expected_points = [(1000, 11.1)]
+    assert expected_points == data_points
+
+
+@skip_ifmodversion_lt("1.12.0", "timeseries")
+def test_decrby_with_insertion_filters(client):
+    assert 1000 == client.ts().decrby(
+        "time-series-1",
+        1.0,
+        timestamp=1000,
+        duplicate_policy="last",
+        ignore_max_time_diff=5,
+        ignore_max_val_diff=10.0,
+    )
+
+    assert 1000 == client.ts().decrby("time-series-1", 3.0, timestamp=1000)
+
+    data_points = client.ts().range("time-series-1", "-", "+")
+    expected_points = [(1000, -1.0)]
+    assert expected_points == data_points
+
+    assert 1000 == client.ts().decrby("time-series-1", 10.1, timestamp=1000)
+
+    data_points = client.ts().range("time-series-1", "-", "+")
+    expected_points = [(1000, -11.1)]
+    assert expected_points == data_points