Skip to content

Add type hints for list commands #1917

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Feb 3, 2022
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 39 additions & 30 deletions redis/commands/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import hashlib
import time
import warnings
from typing import List, Optional
from typing import List, Optional, Union

from redis.exceptions import ConnectionError, DataError, NoScriptError, RedisError

Expand Down Expand Up @@ -1864,7 +1864,7 @@ class ListCommands:
see: https://redis.io/topics/data-types#lists
"""

def blpop(self, keys, timeout=0):
def blpop(self, keys: List, timeout: Optional[int] = 0) -> List:
"""
LPOP a value off of the first non-empty list
named in the ``keys`` list.
Expand All @@ -1883,7 +1883,7 @@ def blpop(self, keys, timeout=0):
keys.append(timeout)
return self.execute_command("BLPOP", *keys)

def brpop(self, keys, timeout=0):
def brpop(self, keys: List, timeout: Optional[int] = 0) -> List:
"""
RPOP a value off of the first non-empty list
named in the ``keys`` list.
Expand All @@ -1902,7 +1902,9 @@ def brpop(self, keys, timeout=0):
keys.append(timeout)
return self.execute_command("BRPOP", *keys)

def brpoplpush(self, src, dst, timeout=0):
def brpoplpush(
self, src: str, dst: str, timeout: Optional[int] = 0
) -> Optional[str]:
"""
Pop a value off the tail of ``src``, push it on the head of ``dst``
and then return it.
Expand Down Expand Up @@ -1957,7 +1959,7 @@ def lmpop(

return self.execute_command("LMPOP", *args)

def lindex(self, name, index):
def lindex(self, name: str, index: int) -> Optional[str]:
"""
Return the item from list ``name`` at position ``index``

Expand All @@ -1968,7 +1970,7 @@ def lindex(self, name, index):
"""
return self.execute_command("LINDEX", name, index)

def linsert(self, name, where, refvalue, value):
def linsert(self, name: str, where: str, refvalue: str, value: str) -> int:
"""
Insert ``value`` in list ``name`` either immediately before or after
[``where``] ``refvalue``
Expand All @@ -1980,15 +1982,15 @@ def linsert(self, name, where, refvalue, value):
"""
return self.execute_command("LINSERT", name, where, refvalue, value)

def llen(self, name):
def llen(self, name: str) -> int:
"""
Return the length of the list ``name``

For more information check https://redis.io/commands/llen
"""
return self.execute_command("LLEN", name)

def lpop(self, name, count=None):
def lpop(self, name: str, count: Optional[int] = None) -> Union[str, List, None]:
"""
Removes and returns the first elements of the list ``name``.

Expand All @@ -2003,23 +2005,23 @@ def lpop(self, name, count=None):
else:
return self.execute_command("LPOP", name)

def lpush(self, name, *values):
def lpush(self, name: str, *values: List) -> int:
"""
Push ``values`` onto the head of the list ``name``

For more information check https://redis.io/commands/lpush
"""
return self.execute_command("LPUSH", name, *values)

def lpushx(self, name, *values):
def lpushx(self, name: str, *values: List) -> int:
"""
Push ``value`` onto the head of the list ``name`` if ``name`` exists

For more information check https://redis.io/commands/lpushx
"""
return self.execute_command("LPUSHX", name, *values)

def lrange(self, name, start, end):
def lrange(self, name: str, start: int, end: int) -> List:
"""
Return a slice of the list ``name`` between
position ``start`` and ``end``
Expand All @@ -2031,7 +2033,7 @@ def lrange(self, name, start, end):
"""
return self.execute_command("LRANGE", name, start, end)

def lrem(self, name, count, value):
def lrem(self, name: str, count: int, value: str) -> int:
"""
Remove the first ``count`` occurrences of elements equal to ``value``
from the list stored at ``name``.
Expand All @@ -2045,15 +2047,15 @@ def lrem(self, name, count, value):
"""
return self.execute_command("LREM", name, count, value)

def lset(self, name, index, value):
def lset(self, name: str, index: int, value: str) -> str:
"""
Set ``position`` of list ``name`` to ``value``
Set element at ``index`` of list ``name`` to ``value``

For more information check https://redis.io/commands/lset
"""
return self.execute_command("LSET", name, index, value)

def ltrim(self, name, start, end):
def ltrim(self, name: str, start: int, end: int) -> str:
"""
Trim the list ``name``, removing all values not within the slice
between ``start`` and ``end``
Expand All @@ -2065,7 +2067,7 @@ def ltrim(self, name, start, end):
"""
return self.execute_command("LTRIM", name, start, end)

def rpop(self, name, count=None):
def rpop(self, name: str, count: Optional[int] = None) -> Union[str, List, None]:
"""
Removes and returns the last elements of the list ``name``.

Expand All @@ -2080,7 +2082,7 @@ def rpop(self, name, count=None):
else:
return self.execute_command("RPOP", name)

def rpoplpush(self, src, dst):
def rpoplpush(self, src: str, dst: str) -> str:
"""
RPOP a value off of the ``src`` list and atomically LPUSH it
on to the ``dst`` list. Returns the value.
Expand All @@ -2089,23 +2091,30 @@ def rpoplpush(self, src, dst):
"""
return self.execute_command("RPOPLPUSH", src, dst)

def rpush(self, name, *values):
def rpush(self, name: str, *values: List) -> int:
"""
Push ``values`` onto the tail of the list ``name``

For more information check https://redis.io/commands/rpush
"""
return self.execute_command("RPUSH", name, *values)

def rpushx(self, name, value):
def rpushx(self, name: str, value: str) -> int:
"""
Push ``value`` onto the tail of the list ``name`` if ``name`` exists

For more information check https://redis.io/commands/rpushx
"""
return self.execute_command("RPUSHX", name, value)

def lpos(self, name, value, rank=None, count=None, maxlen=None):
def lpos(
self,
name: str,
value: str,
rank: Optional[int] = None,
count: Optional[int] = None,
maxlen: Optional[int] = None,
) -> Union[str, List, None]:
"""
Get position of ``value`` within the list ``name``

Expand Down Expand Up @@ -2145,16 +2154,16 @@ def lpos(self, name, value, rank=None, count=None, maxlen=None):

def sort(
self,
name,
start=None,
num=None,
by=None,
get=None,
desc=False,
alpha=False,
store=None,
groups=False,
):
name: str,
start: Optional[int] = None,
num: Optional[int] = None,
by: Optional[str] = None,
get: Optional[List[str]] = None,
desc: bool = False,
alpha: bool = False,
store: Optional[str] = None,
groups: Optional[bool] = False,
) -> Union[List, int]:
"""
Sort and return the list, set or sorted set at ``name``.

Expand Down