Update Type Hints for List Command Parameters from str to KeyT

redis/commands/core.py

@@ -2792,7 +2792,7 @@ def brpop(
         return self.execute_command("BRPOP", *keys)
 
     def brpoplpush(
-        self, src: str, dst: str, timeout: Optional[Number] = 0
+        self, src: KeyT, dst: KeyT, timeout: Optional[Number] = 0
     ) -> Union[Awaitable[Optional[str]], Optional[str]]:
         """
         Pop a value off the tail of ``src``, push it on the head of ``dst``
@@ -2849,7 +2849,7 @@ def lmpop(
         return self.execute_command("LMPOP", *cmd_args)
 
     def lindex(
-        self, name: str, index: int
+        self, name: KeyT, index: int
     ) -> Union[Awaitable[Optional[str]], Optional[str]]:
         """
         Return the item from list ``name`` at position ``index``
@@ -2862,7 +2862,7 @@ def lindex(
         return self.execute_command("LINDEX", name, index, keys=[name])
 
     def linsert(
-        self, name: str, where: str, refvalue: str, value: str
+        self, name: KeyT, where: str, refvalue: str, value: str
     ) -> Union[Awaitable[int], int]:
         """
         Insert ``value`` in list ``name`` either immediately before or after
@@ -2875,7 +2875,7 @@ def linsert(
         """
         return self.execute_command("LINSERT", name, where, refvalue, value)
 
-    def llen(self, name: str) -> Union[Awaitable[int], int]:
+    def llen(self, name: KeyT) -> Union[Awaitable[int], int]:
         """
         Return the length of the list ``name``
 
@@ -2885,7 +2885,7 @@ def llen(self, name: str) -> Union[Awaitable[int], int]:
 
     def lpop(
         self,
-        name: str,
+        name: KeyT,
         count: Optional[int] = None,
     ) -> Union[Awaitable[Union[str, List, None]], Union[str, List, None]]:
         """
@@ -2902,23 +2902,23 @@ def lpop(
         else:
             return self.execute_command("LPOP", name)
 
-    def lpush(self, name: str, *values: FieldT) -> Union[Awaitable[int], int]:
+    def lpush(self, name: KeyT, *values: FieldT) -> Union[Awaitable[int], int]:
         """
         Push ``values`` onto the head of the list ``name``
 
         For more information, see https://redis.io/commands/lpush
         """
         return self.execute_command("LPUSH", name, *values)
 
-    def lpushx(self, name: str, *values: FieldT) -> Union[Awaitable[int], int]:
+    def lpushx(self, name: KeyT, *values: FieldT) -> Union[Awaitable[int], int]:
         """
         Push ``value`` onto the head of the list ``name`` if ``name`` exists
 
         For more information, see https://redis.io/commands/lpushx
         """
         return self.execute_command("LPUSHX", name, *values)
 
-    def lrange(self, name: str, start: int, end: int) -> Union[Awaitable[list], list]:
+    def lrange(self, name: KeyT, start: int, end: int) -> Union[Awaitable[list], list]:
         """
         Return a slice of the list ``name`` between
         position ``start`` and ``end``
@@ -2930,7 +2930,7 @@ def lrange(self, name: str, start: int, end: int) -> Union[Awaitable[list], list
         """
         return self.execute_command("LRANGE", name, start, end, keys=[name])
 
-    def lrem(self, name: str, count: int, value: str) -> Union[Awaitable[int], int]:
+    def lrem(self, name: KeyT, count: int, value: str) -> Union[Awaitable[int], int]:
         """
         Remove the first ``count`` occurrences of elements equal to ``value``
         from the list stored at ``name``.
@@ -2944,15 +2944,15 @@ def lrem(self, name: str, count: int, value: str) -> Union[Awaitable[int], int]:
         """
         return self.execute_command("LREM", name, count, value)
 
-    def lset(self, name: str, index: int, value: str) -> Union[Awaitable[str], str]:
+    def lset(self, name: KeyT, index: int, value: str) -> Union[Awaitable[str], str]:
         """
         Set element at ``index`` of list ``name`` to ``value``
 
         For more information, see https://redis.io/commands/lset
         """
         return self.execute_command("LSET", name, index, value)
 
-    def ltrim(self, name: str, start: int, end: int) -> Union[Awaitable[str], str]:
+    def ltrim(self, name: KeyT, start: int, end: int) -> Union[Awaitable[str], str]:
         """
         Trim the list ``name``, removing all values not within the slice
         between ``start`` and ``end``
@@ -2966,7 +2966,7 @@ def ltrim(self, name: str, start: int, end: int) -> Union[Awaitable[str], str]:
 
     def rpop(
         self,
-        name: str,
+        name: KeyT,
         count: Optional[int] = None,
     ) -> Union[Awaitable[Union[str, List, None]], Union[str, List, None]]:
         """
@@ -2983,7 +2983,7 @@ def rpop(
         else:
             return self.execute_command("RPOP", name)
 
-    def rpoplpush(self, src: str, dst: str) -> Union[Awaitable[str], str]:
+    def rpoplpush(self, src: KeyT, dst: KeyT) -> Union[Awaitable[str], str]:
         """
         RPOP a value off of the ``src`` list and atomically LPUSH it
         on to the ``dst`` list.  Returns the value.
@@ -2992,15 +2992,15 @@ def rpoplpush(self, src: str, dst: str) -> Union[Awaitable[str], str]:
         """
         return self.execute_command("RPOPLPUSH", src, dst)
 
-    def rpush(self, name: str, *values: FieldT) -> Union[Awaitable[int], int]:
+    def rpush(self, name: KeyT, *values: FieldT) -> Union[Awaitable[int], int]:
         """
         Push ``values`` onto the tail of the list ``name``
 
         For more information, see https://redis.io/commands/rpush
         """
         return self.execute_command("RPUSH", name, *values)
 
-    def rpushx(self, name: str, *values: str) -> Union[Awaitable[int], int]:
+    def rpushx(self, name: KeyT, *values: str) -> Union[Awaitable[int], int]:
         """
         Push ``value`` onto the tail of the list ``name`` if ``name`` exists
 
@@ -3010,7 +3010,7 @@ def rpushx(self, name: str, *values: str) -> Union[Awaitable[int], int]:
 
     def lpos(
         self,
-        name: str,
+        name: KeyT,
         value: str,
         rank: Optional[int] = None,
         count: Optional[int] = None,
@@ -3055,7 +3055,7 @@ def lpos(
 
     def sort(
         self,
-        name: str,
+        name: KeyT,
         start: Optional[int] = None,
         num: Optional[int] = None,
         by: Optional[str] = None,

tests/test_commands.py

@@ -2892,6 +2892,8 @@ def test_lrange(self, r):
         assert r.lrange("a", 0, 2) == [b"1", b"2", b"3"]
         assert r.lrange("a", 2, 10) == [b"3", b"4", b"5"]
         assert r.lrange("a", 0, -1) == [b"1", b"2", b"3", b"4", b"5"]
+        r.rpush(bytes(345), "12", "22", "32", "42", "52")
+        assert r.lrange(bytes(345), 0, 0) == [b"12"]
 
     def test_lrem(self, r):
         r.rpush("a", "Z", "b", "Z", "Z", "c", "Z", "Z")
@@ -2984,6 +2986,97 @@ def test_rpushx(self, r):
         assert r.rpushx("a", "4") == 4
         assert r.lrange("a", 0, -1) == [b"1", b"2", b"3", b"4"]
 
+    def test_lists_with_byte_keys(self, r):
+        r.rpush(b"b", b"1", b"2", b"3")
+        assert r.lrange(b"b", 0, -1) == [b"1", b"2", b"3"]
+        # LPOS command with byte keys
+        assert r.lpos(b"b", b"2") == 1
+        assert r.lpos(b"b", b"2", rank=1) == 1
+        assert r.lpos(b"b", b"2", rank=2) is None
+        # LCS command with byte keys
+        r.set(b"key1", b"ohmytext")
+        r.set(b"key2", b"mynewtext")
+        assert r.lcs(b"key1", b"key2") == b"mytext"
+        # TYPE command with byte keys
+        assert r.type(b"b") == b"list"
+        assert r.type(b"key1") == b"string"
+        # SCAN command with byte keys
+        r.set(b"scan_key1", b"value1")
+        r.set(b"scan_key2", b"value2")
+        cursor, keys = r.scan(match=b"scan_key*")
+        assert cursor == 0
+        assert set(keys) == {b"scan_key1", b"scan_key2"}
+        # PEXPIRETIME command with byte keys
+        r.set(b"expire_key", b"value")
+        r.pexpire(b"expire_key", 10000)
+        pexpiretime = r.pexpiretime(b"expire_key")
+        assert pexpiretime > 0
+        # LMOVE command with byte keys (src and dest)
+        r.rpush(b"list_src", b"a", b"b", b"c")
+        r.rpush(b"list_dest", b"x")
+        moved = r.lmove(b"list_src", b"list_dest", src=b"LEFT", dest=b"RIGHT")
+        assert moved == b"a"
+        assert r.lrange(b"list_dest", 0, -1) == [b"x", b"a"]
+        # SMOVE command with byte keys (src and dst)
+        r.sadd(b"set_src", b"member1", b"member2")
+        r.sadd(b"set_dest", b"member3")
+        assert r.smove(b"set_src", b"set_dest", b"member1") == 1
+        assert b"member1" in r.smembers(b"set_dest")
+
+    def test_lists_with_memoryview_keys(self, r):
+        # Create memoryview objects for key names
+        mv_b = memoryview(b"b")
+        mv_key1 = memoryview(b"key1")
+        mv_key2 = memoryview(b"key2")
+        mv_scan_key1 = memoryview(b"scan_key1")
+        mv_scan_key2 = memoryview(b"scan_key2")
+        mv_expire_key = memoryview(b"expire_key")
+        mv_list_src = memoryview(b"list_src")
+        mv_list_dest = memoryview(b"list_dest")
+        mv_set_src = memoryview(b"set_src")
+        mv_set_dest = memoryview(b"set_dest")
+
+        r.rpush(mv_b, b"1", b"2", b"3")
+        assert r.lrange(mv_b, 0, -1) == [b"1", b"2", b"3"]
+        # LPOS command with memoryview keys
+        assert r.lpos(mv_b, b"2") == 1
+        assert r.lpos(mv_b, b"2", rank=1) == 1
+        assert r.lpos(mv_b, b"2", rank=2) is None
+        # LCS command with memoryview keys
+        r.set(mv_key1, b"ohmytext")
+        r.set(mv_key2, b"mynewtext")
+        assert r.lcs(mv_key1, mv_key2) == b"mytext"
+        # TYPE command with memoryview keys
+        assert r.type(mv_b) == b"list"
+        assert r.type(mv_key1) == b"string"
+        # SCAN command with memoryview keys
+        r.set(mv_scan_key1, b"value1")
+        r.set(mv_scan_key2, b"value2")
+        cursor, keys = r.scan(match=memoryview(b"scan_key*"))
+        assert cursor == 0
+        assert set(keys) == {b"scan_key1", b"scan_key2"}
+        # PEXPIRETIME command with memoryview keys
+        r.set(mv_expire_key, b"value")
+        r.pexpire(mv_expire_key, 10000)
+        pexpiretime = r.pexpiretime(mv_expire_key)
+        assert pexpiretime > 0
+        # LMOVE command with memoryview keys (src and dest)
+        r.rpush(mv_list_src, b"a", b"b", b"c")
+        r.rpush(mv_list_dest, b"x")
+        moved = r.lmove(
+            mv_list_src,
+            mv_list_dest,
+            src=memoryview(b"LEFT"),
+            dest=memoryview(b"RIGHT"),
+        )
+        assert moved == b"a"
+        assert r.lrange(mv_list_dest, 0, -1) == [b"x", b"a"]
+        # SMOVE command with memoryview keys (src and dst)
+        r.sadd(mv_set_src, b"member1", b"member2")
+        r.sadd(mv_set_dest, b"member3")
+        assert r.smove(mv_set_src, mv_set_dest, b"member1") == 1
+        assert b"member1" in r.smembers(mv_set_dest)
+
     # SCAN COMMANDS
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("2.8.0")