Add support for certain LATENCY commands

redis/cluster.py

@@ -265,6 +265,9 @@ class AbstractRedisCluster:
                 "READWRITE",
                 "TIME",
                 "GRAPH.CONFIG",
+                "LATENCY HISTORY",
+                "LATENCY LATEST",
+                "LATENCY RESET",
             ],
             DEFAULT_NODE,
         ),

redis/commands/core.py

@@ -1151,6 +1151,30 @@ def latency_histogram(self, *args):
             "LATENCY HISTOGRAM is intentionally not implemented in the client."
         )
 
+    def latency_history(self, event: str) -> ResponseT:
+        """
+        Returns the raw data of the ``event``'s latency spikes time series.
+
+        For more information see https://redis.io/commands/latency-history
+        """
+        return self.execute_command("LATENCY HISTORY", event)
+
+    def latency_latest(self) -> ResponseT:
+        """
+        Reports the latest latency events logged.
+
+        For more information see https://redis.io/commands/latency-latest
+        """
+        return self.execute_command("LATENCY LATEST")
+
+    def latency_reset(self, *events: str) -> ResponseT:
+        """
+        Resets the latency spikes time series of all, or only some, events.
+
+        For more information see https://redis.io/commands/latency-reset
+        """
+        return self.execute_command("LATENCY RESET", *events)
+
     def ping(self, **kwargs) -> ResponseT:
         """
         Ping the Redis server

tests/test_commands.py

@@ -4527,16 +4527,23 @@ def test_latency_histogram_not_implemented(self, r: redis.Redis):
         with pytest.raises(NotImplementedError):
             r.latency_histogram()
 
-    @skip_if_server_version_lt("7.0.0")
     def test_latency_graph_not_implemented(self, r: redis.Redis):
         with pytest.raises(NotImplementedError):
             r.latency_graph()
 
-    @skip_if_server_version_lt("7.0.0")
     def test_latency_doctor_not_implemented(self, r: redis.Redis):
         with pytest.raises(NotImplementedError):
             r.latency_doctor()
 
+    def test_latency_history(self, r: redis.Redis):
+        assert r.latency_history("command") == []
+
+    def test_latency_latest(self, r: redis.Redis):
+        assert r.latency_latest() == []
+
+    def test_latency_reset(self, r: redis.Redis):
+        assert r.latency_reset() == 0
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("4.0.0")
     @skip_if_redis_enterprise()