Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_zset(value):
if isinstance(value, set):
zset = ZSet()
for item in value:
zset[item] = 1.0
return zset
elif isinstance(value, ZSet):
return value
else:
raise redis.ResponseError(WRONGTYPE_MSG)
elif casematch(arg, b'aggregate') and i + 1 < len(args):
aggregate = casenorm(args[i + 1])
if aggregate not in (b'sum', b'min', b'max'):
raise redis.ResponseError(SYNTAX_ERROR_MSG)
i += 2
else:
raise redis.ResponseError(SYNTAX_ERROR_MSG)
out_members = set(sets[0])
for s in sets[1:]:
if func == 'ZUNIONSTORE':
out_members |= set(s)
else:
out_members.intersection_update(s)
out = ZSet()
for s, w in zip(sets, weights):
for member, score in s.items():
score *= w
if math.isnan(score):
score = 0.0
if member not in out_members:
continue
if member in out:
old = out[member]
if aggregate == b'sum':
score += old
if math.isnan(score):
score = 0.0
elif aggregate == b'max':
score = max(old, score)
elif aggregate == b'min':
@command((Key(ZSet), bytes), (bytes,))
def zrem(self, key, *members):
old_size = len(key.value)
for member in members:
key.value.discard(member)
deleted = old_size - len(key.value)
if deleted:
key.updated()
return deleted
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