Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def client_getname(self, encoding=_NOTSET):
"""Get the current connection name."""
return self.execute(b'CLIENT', b'GETNAME', encoding=encoding)
def zrange(self, key, start=0, stop=-1, withscores=False,
encoding=_NOTSET):
"""Return a range of members in a sorted set, by index.
:raises TypeError: if start is not int
:raises TypeError: if stop is not int
"""
if not isinstance(start, int):
raise TypeError("start argument must be int")
if not isinstance(stop, int):
raise TypeError("stop argument must be int")
if withscores:
args = [b'WITHSCORES']
else:
args = []
fut = self.execute(b'ZRANGE', key, start, stop, *args,
encoding=encoding)
if withscores:
def brpop(self, key, *keys, timeout=0, encoding=_NOTSET):
"""Remove and get the last element in a list, or block until one
is available.
:raises TypeError: if timeout is not int
:raises ValueError: if timeout is less than 0
"""
if not isinstance(timeout, int):
raise TypeError("timeout argument must be int")
if timeout < 0:
raise ValueError("timeout must be greater equal 0")
args = keys + (timeout,)
return self.execute(b'BRPOP', key, *args, encoding=encoding)
def blpop(self, key, *keys, timeout=0, encoding=_NOTSET):
"""Remove and get the first element in a list, or block until
one is available.
:raises TypeError: if timeout is not int
:raises ValueError: if timeout is less than 0
"""
if not isinstance(timeout, int):
raise TypeError("timeout argument must be int")
if timeout < 0:
raise ValueError("timeout must be greater equal 0")
args = keys + (timeout,)
return self.execute(b'BLPOP', key, *args, encoding=encoding)
def execute(self, command, *args, encoding=_NOTSET):
"""Executes redis command and returns Future waiting for the answer.
Raises:
* TypeError if any of args can not be encoded as bytes.
* ReplyError on redis '-ERR' resonses.
* ProtocolError when response can not be decoded meaning connection
is broken.
"""
if self._reader is None or self._reader.at_eof():
raise ConnectionClosedError("Connection closed or corrupted")
if command is None:
raise TypeError("command must not be None")
if None in set(args):
raise TypeError("args must not contain None")
command = command.upper().strip()
is_pubsub = command in _PUBSUB_COMMANDS
def zrangebyscore(self, key, min=float('-inf'), max=float('inf'),
withscores=False, offset=None, count=None,
*, exclude=None, encoding=_NOTSET):
"""Return a range of members in a sorted set, by score.
:raises TypeError: if min or max is not float or int
:raises TypeError: if both offset and count are not specified
:raises TypeError: if offset is not int
:raises TypeError: if count is not int
"""
if not isinstance(min, (int, float)):
raise TypeError("min argument must be int or float")
if not isinstance(max, (int, float)):
raise TypeError("max argument must be int or float")
if (offset is not None and count is None) or \
(count is not None and offset is None):
raise TypeError("offset and count must both be specified")
if offset is not None and not isinstance(offset, int):
def ping(self, message=_NOTSET, *, encoding=_NOTSET):
"""Ping the server.
Accept optional echo message.
"""
if message is not _NOTSET:
args = (message,)
else:
args = ()
return self.execute('PING', *args, encoding=encoding)
async def hgetall_async(key: str, *, decode: bool = True) -> dict:
await _set_redis_client_async_if_necessary()
return await async_redis_client.hgetall(key, encoding="utf8" if decode else aioredis.util._NOTSET)