Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def tearDown(self):
keys_to_delete = []
for prefix in {Base._RANDOM_KEY_PREFIX, self._TEST_KEY_PREFIX}:
pattern = prefix + '*'
keys = self.redis.keys(pattern=pattern)
keys = (key.decode('utf-8') for key in keys)
keys_to_delete.extend(keys)
if keys_to_delete:
self.redis.delete(*keys_to_delete)
super().tearDown()
def test_init_with_wrong_type_maxlen(self):
with unittest.mock.patch.object(Base, '__del__') as delete, \
self.assertRaises(TypeError):
delete.return_value = None
RedisDeque(maxlen='2')
# set.py #
# #
# Copyright © 2015-2020, Rajiv Bakulesh Shah, original author. #
# All rights reserved. #
# --------------------------------------------------------------------------- #
import collections.abc
import itertools
from .base import Base
from .base import Iterable
from .exceptions import KeyExistsError
class RedisSet(Base, Iterable, collections.abc.MutableSet):
'Redis-backed container compatible with Python sets.'
def __init__(self, iterable=tuple(), *, redis=None, key=None):
'Initialize a RedisSet. O(n)'
super().__init__(iterable, redis=redis, key=key)
with self._watch(iterable):
self._populate(iterable)
def _populate(self, iterable=tuple()):
encoded_values = {self._encode(value) for value in iterable}
if encoded_values:
if self.redis.exists(self.key):
raise KeyExistsError(self.redis, self.key)
else:
self.redis.multi()
self.redis.sadd(self.key, *encoded_values)
def _context_managers(self, *others):
redises = collections.defaultdict(list)
for container in itertools.chain((self,), others):
if isinstance(container, Base):
redises[container.redis].append(container)
for containers in redises.values():
keys = (container.key for container in containers)
yield containers[0]._watch_keys(*keys)
# Copyright © 2015-2020, Rajiv Bakulesh Shah, original author. #
# All rights reserved. #
# --------------------------------------------------------------------------- #
import collections.abc
import functools
import itertools
from redis import ResponseError
from .base import Base
from .exceptions import KeyExistsError
class RedisList(Base, collections.abc.MutableSequence):
'Redis-backed container compatible with Python lists.'
def _raise_on_error(func):
@functools.wraps(func)
def wrap(*args, **kwargs):
try:
return func(*args, **kwargs)
except ResponseError:
raise IndexError('list assignment index out of range')
return wrap
def _slice_to_indices(self, slice_or_index):
try:
start = slice_or_index.start or 0
stop = slice_or_index.stop or len(self)
step = slice_or_index.step or 1
# --------------------------------------------------------------------------- #
# hyper.py #
# #
# Copyright © 2015-2020, Rajiv Bakulesh Shah, original author. #
# All rights reserved. #
# --------------------------------------------------------------------------- #
from .base import Base
class HyperLogLog(Base):
'''Redis-backed HyperLogLog with a Pythonic API.
Wikipedia article:
https://en.wikipedia.org/wiki/HyperLogLog
antirez's blog post:
http://antirez.com/news/75
'''
def __init__(self, iterable=frozenset(), *, redis=None, key=None):
'''Initialize a HyperLogLog. O(n)
Here, n is the number of elements in iterable that you want to insert
into this HyperLogLog.
'''
super().__init__(redis=redis, key=key)
# #
# Copyright © 2015-2020, Rajiv Bakulesh Shah, original author. #
# All rights reserved. #
# --------------------------------------------------------------------------- #
import collections.abc
import contextlib
import itertools
from .base import Base
from .base import Iterable
from .exceptions import KeyExistsError
class RedisDict(Base, Iterable, collections.abc.MutableMapping):
'Redis-backed container compatible with Python dicts.'
def __init__(self, iterable=tuple(), *, redis=None, key=None, **kwargs):
'Initialize a RedisDict. O(n)'
super().__init__(redis=redis, key=key, **kwargs)
if iterable or kwargs:
with self._watch(iterable):
if self.redis.exists(self.key):
raise KeyExistsError(self.redis, self.key)
else:
self._populate(iterable, **kwargs)
def _populate(self, iterable=tuple(), **kwargs):
to_set = {}
with contextlib.suppress(AttributeError):
iterable = iterable.items()
# bloom.py #
# #
# Copyright © 2015-2020, Rajiv Bakulesh Shah, original author. #
# All rights reserved. #
# --------------------------------------------------------------------------- #
import itertools
import math
import mmh3
from .base import Base
class BloomFilter(Base):
'''Redis-backed Bloom filter with an API similar to Python sets.
Bloom filters are a powerful data structure that help you to answer the
question, "Have I seen this element before?" but not the question, "What
are all of the elements that I've seen before?" So think of Bloom filters
as Python sets that you can add elements to and use to test element
membership, but that you can't iterate through or get elements back out of.
Bloom filters are probabilistic, which means that they can sometimes
generate false positives (as in, they may report that you've seen a
particular element before even though you haven't). But they will never
generate false negatives (so every time that they report that you haven't
seen a particular element before, you really must never have seen it). You
can tune your acceptable false positive probability, though at the expense
of the storage size and the element insertion/lookup time of your Bloom
filter.