Add BITFIELD_RO (#2340)

redis/commands/core.py

@@ -1504,6 +1504,29 @@ def bitfield(
         """
         return BitFieldOperation(self, key, default_overflow=default_overflow)
 
+    def bitfield_ro(
+        self: Union["Redis", "AsyncRedis"],
+        key: KeyT,
+        encoding: str,
+        offset: BitfieldOffsetT,
+        items: Optional[list] = None,
+    ) -> ResponseT:
+        """
+        Return an array of the specified bitfield values
+        where the first value is found using ``encoding`` and ``offset``
+        parameters and remaining values are result of corresponding
+        encoding/offset pairs in optional list ``items``
+        Read-only variant of the BITFIELD command.
+
+        For more information see https://redis.io/commands/bitfield_ro
+        """
+        params = [key, "GET", encoding, offset]
+
+        items = items or []
+        for encoding, offset in items:
+            params.extend(["GET", encoding, offset])
+        return self.execute_command("BITFIELD_RO", *params)
+
     def bitop(self, operation: str, dest: KeyT, *keys: KeyT) -> ResponseT:
         """
         Perform a bitwise operation using ``operation`` between ``keys`` and

tests/test_asyncio/test_commands.py

@@ -2953,6 +2953,19 @@ async def test_bitfield_operations(self, r: redis.Redis):
         )
         assert resp == [0, None, 255]
 
+    @skip_if_server_version_lt("6.0.0")
+    async def test_bitfield_ro(self, r: redis.Redis):
+        bf = r.bitfield("a")
+        resp = await bf.set("u8", 8, 255).execute()
+        assert resp == [0]
+
+        resp = await r.bitfield_ro("a", "u8", 0)
+        assert resp == [0]
+
+        items = [("u4", 8), ("u4", 12), ("u4", 13)]
+        resp = await r.bitfield_ro("a", "u8", 0, items)
+        assert resp == [0, 15, 15, 14]
+
     @skip_if_server_version_lt("4.0.0")
     async def test_memory_stats(self, r: redis.Redis):
         # put a key into the current db to make sure that "db.<current-db>"

tests/test_commands.py

@@ -4438,6 +4438,19 @@ def test_bitfield_operations(self, r):
         )
         assert resp == [0, None, 255]
 
+    @skip_if_server_version_lt("6.0.0")
+    def test_bitfield_ro(self, r: redis.Redis):
+        bf = r.bitfield("a")
+        resp = bf.set("u8", 8, 255).execute()
+        assert resp == [0]
+
+        resp = r.bitfield_ro("a", "u8", 0)
+        assert resp == [0]
+
+        items = [("u4", 8), ("u4", 12), ("u4", 13)]
+        resp = r.bitfield_ro("a", "u8", 0, items)
+        assert resp == [0, 15, 15, 14]
+
     @skip_if_server_version_lt("4.0.0")
     def test_memory_help(self, r):
         with pytest.raises(NotImplementedError):