Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _apply_withscores(items, withscores):
if withscores:
out = []
for item in items:
out.append(item[1])
out.append(Float.encode(item[0], False))
else:
out = [item[1] for item in items]
return out
def hincrbyfloat(self, key, field, amount):
c = Float.decode(key.value.get(field, b'0')) + Float.decode(amount)
if not isfinite(c):
raise redis.ResponseError(NONFINITE_MSG)
encoded = Float.encode(c, True)
key.value[field] = encoded
key.updated()
return encoded
@command((Key(ZSet), Float, bytes))
def zincrby(self, key, increment, member):
# Can't just default the old score to 0.0, because in IEEE754, adding
# 0.0 to something isn't a nop (e.g. 0.0 + -0.0 == 0.0).
try:
score = key.value.get(member, None) + increment
except TypeError:
score = increment
if math.isnan(score):
raise redis.ResponseError(SCORE_NAN_MSG)
key.value[member] = score
key.updated()
return Float.encode(score, False)
raise redis.ResponseError(cls.DECODE_ERROR)
@classmethod
def encode(cls, value, humanfriendly):
if math.isinf(value):
return six.ensure_binary(str(value))
elif humanfriendly:
# Algorithm from ld2string in redis
out = '{:.17f}'.format(value)
out = re.sub(r'(?:\.)?0+$', '', out)
return six.ensure_binary(out)
else:
return six.ensure_binary('{:.17g}'.format(value))
class SortFloat(Float):
DECODE_ERROR = INVALID_SORT_FLOAT_MSG
@classmethod
def decode(cls, value):
return super(SortFloat, cls).decode(
value, allow_leading_whitespace=True, allow_empty=True, crop_null=True)
class ScoreTest(object):
"""Argument converter for sorted set score endpoints."""
def __init__(self, value, exclusive=False):
self.value = value
self.exclusive = exclusive
@classmethod
def decode(cls, value):