Add support for BLMPOP (#1849)

* Add support for BLMPOP

* add type hints

* fix test

* fix comment

* fix pr comment

* delete count check

* change numkeys

* linters

* mark test as onlynoncluster

redis/commands/core.py

@@ -1917,6 +1917,27 @@ def brpoplpush(self, src, dst, timeout=0):
             timeout = 0
         return self.execute_command("BRPOPLPUSH", src, dst, timeout)
 
+    def blmpop(
+        self,
+        timeout: float,
+        numkeys: 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, numkeys, *args, direction, "COUNT", count]
+
+        return self.execute_command("BLMPOP", *args)
+
     def lmpop(
         self,
         num_keys: int,

tests/test_commands.py

@@ -1480,6 +1480,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
+
     @pytest.mark.onlynoncluster
     # @skip_if_server_version_lt("7.0.0") turn on after redis 7 release
     def test_lmpop(self, unstable_r):