Add support for BLMPOP

redis/commands/core.py

@@ -2,6 +2,7 @@
 import hashlib
 import time
 import warnings
+from typing import List, Optional
 
 from redis.exceptions import ConnectionError, DataError, NoScriptError, RedisError
 
@@ -1926,6 +1927,29 @@ def brpoplpush(self, src, dst, timeout=0):
             timeout = 0
         return self.execute_command("BRPOPLPUSH", src, dst, timeout)
 
+    def blmpop(
+        self,
+        timeout: float,
+        num_keys: int,
+        *args: List[str],
+        direction: str,
+        count: Optional[int] = 1,
+    ) -> Optional[list]:
+        """
+        Pop ``count`` values (default 1) from first non-empty in the list
+        of provided key names.
+
+        When all lists are empty this command blocks the connection until another
+        client pushes to it or until the timeout, timeout of 0 blocks indefinitely
+
+        For more information check https://redis.io/commands/blmpop
+        """
+        args = [timeout, num_keys, *args, direction]
+        if count != 1:
+            args.extend(["COUNT", count])
+
+        return self.execute_command("BLMPOP", *args)
+
     def lindex(self, name, index):
         """
         Return the item from list ``name`` at position ``index``

tests/test_commands.py

@@ -1473,6 +1473,18 @@ def test_brpoplpush_empty_string(self, r):
         r.rpush("a", "")
         assert r.brpoplpush("a", "b") == b""
 
+    @pytest.mark.onlynoncluster
+    # @skip_if_server_version_lt("7.0.0") turn on after redis 7 release
+    def test_blmpop(self, unstable_r):
+        unstable_r.rpush("a", "1", "2", "3", "4", "5")
+        res = [b"a", [b"1", b"2"]]
+        assert unstable_r.blmpop(1, "2", "b", "a", direction="LEFT", count=2) == res
+        with pytest.raises(TypeError):
+            unstable_r.blmpop(1, "2", "b", "a", count=2)
+        unstable_r.rpush("b", "6", "7", "8", "9")
+        assert unstable_r.blmpop(0, "2", "b", "a", direction="LEFT") == [b"b", [b"6"]]
+        assert unstable_r.blmpop(1, "2", "foo", "bar", direction="RIGHT") is None
+
     def test_lindex(self, r):
         r.rpush("a", "1", "2", "3")
         assert r.lindex("a", "0") == b"1"