Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def type(self, key):
if key.value is None:
return SimpleString(b'none')
elif isinstance(key.value, bytes):
return SimpleString(b'string')
elif isinstance(key.value, list):
return SimpleString(b'list')
elif isinstance(key.value, set):
return SimpleString(b'set')
elif isinstance(key.value, ZSet):
return SimpleString(b'zset')
elif isinstance(key.value, dict):
return SimpleString(b'hash')
else:
assert False # pragma: nocover
def type(self, key):
if key.value is None:
return SimpleString(b'none')
elif isinstance(key.value, bytes):
return SimpleString(b'string')
elif isinstance(key.value, list):
return SimpleString(b'list')
elif isinstance(key.value, set):
return SimpleString(b'set')
elif isinstance(key.value, ZSet):
return SimpleString(b'zset')
elif isinstance(key.value, dict):
return SimpleString(b'hash')
else:
assert False # pragma: nocover
def _convert_redis_result(self, lua_runtime, result):
if isinstance(result, (bytes, int, long)):
return result
elif isinstance(result, SimpleString):
return lua_runtime.table_from({b"ok": result.value})
elif result is None:
return False
elif isinstance(result, list):
converted = [
self._convert_redis_result(lua_runtime, item)
for item in result
]
return lua_runtime.table_from(converted)
elif isinstance(result, redis.ResponseError):
raise result
else:
raise RuntimeError("Unexpected return type from redis: {}".format(type(result)))
def type(self, key):
if key.value is None:
return SimpleString(b'none')
elif isinstance(key.value, bytes):
return SimpleString(b'string')
elif isinstance(key.value, list):
return SimpleString(b'list')
elif isinstance(key.value, set):
return SimpleString(b'set')
elif isinstance(key.value, ZSet):
return SimpleString(b'zset')
elif isinstance(key.value, dict):
return SimpleString(b'hash')
else:
assert False # pragma: nocover
def type(self, key):
if key.value is None:
return SimpleString(b'none')
elif isinstance(key.value, bytes):
return SimpleString(b'string')
elif isinstance(key.value, list):
return SimpleString(b'list')
elif isinstance(key.value, set):
return SimpleString(b'set')
elif isinstance(key.value, ZSet):
return SimpleString(b'zset')
elif isinstance(key.value, dict):
return SimpleString(b'hash')
else:
assert False # pragma: nocover
def _convert_lua_result(self, result, nested=True):
from lupa import lua_type
if lua_type(result) == 'table':
for key in (b'ok', b'err'):
if key in result:
msg = self._convert_lua_result(result[key])
if not isinstance(msg, bytes):
# TODO: put in a constant for this
raise redis.ResponseError("wrong number or type of arguments")
if key == b'ok':
return SimpleString(msg)
elif nested:
return redis.ResponseError(msg)
else:
raise redis.ResponseError(msg)
# Convert Lua tables into lists, starting from index 1, mimicking the behavior of StrictRedis.
result_list = []
for index in itertools.count(1):
if index not in result:
break
item = result[index]
result_list.append(self._convert_lua_result(item))
return result_list
elif isinstance(result, six.text_type):
return six.ensure_binary(result)
elif isinstance(result, float):
return int(result)
class SimpleString(object):
def __init__(self, value):
assert isinstance(value, bytes)
self.value = value
class NoResponse(object):
"""Returned by pub/sub commands to indicate that no response should be returned"""
pass
OK = SimpleString(b'OK')
QUEUED = SimpleString(b'QUEUED')
PONG = SimpleString(b'PONG')
BGSAVE_STARTED = SimpleString(b'Background saving started')
if six.PY2:
def isfinite(value):
return not math.isinf(value) and not math.isnan(value)
# Not the same as six.byte2int, which takes a bytes in both Python 2+3.
# This takes an integer in Python 3.
def byte_to_int(value):
return ord(value)
else:
def isfinite(value):
return math.isfinite(value)
def byte_to_int(value):
assert isinstance(value, int)
def type(self, key):
if key.value is None:
return SimpleString(b'none')
elif isinstance(key.value, bytes):
return SimpleString(b'string')
elif isinstance(key.value, list):
return SimpleString(b'list')
elif isinstance(key.value, set):
return SimpleString(b'set')
elif isinstance(key.value, ZSet):
return SimpleString(b'zset')
elif isinstance(key.value, dict):
return SimpleString(b'hash')
else:
assert False # pragma: nocover