Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
144 changes: 110 additions & 34 deletions redisdol/base.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,24 @@
"""Base objects for redisdol"""

import operator
import uuid
from functools import wraps
from collections.abc import Iterable
from dol.base import Collection, KvReader, KvPersister

from redis import Redis
from redis.exceptions import ResponseError


def _identity(x):
return x


def _new_sentinel():
"""A value that can't collide with real data (used to address a list position)."""
return b"__redisdol_sentinel__" + uuid.uuid4().hex.encode()


class RedisType:
string = b"string"
list = b"list"
Expand Down Expand Up @@ -226,6 +234,9 @@ class RedisList(MutableSequence):
['Z', 'B', 'C']
"""

#: Number of elements fetched per round trip when iterating.
iter_batch_size = 1000

def __init__(self, source, name, *, encode=None, decode=None):
self._source = source
self._name = name
Expand All @@ -236,36 +247,63 @@ def __len__(self):
return self._source.llen(self._name)

def __getitem__(self, i):
if isinstance(i, int):
v = self._source.lindex(self._name, i)
return v if v is None else self._decode(v)
else: # assume it's a slice, and modify it to conform to inclusive Redis slicing
try:
if i.step is not None:
raise NotImplementedError(
"Slicing is so far only implemented without step"
)
return [
self._decode(v)
for v in self._source.lrange(self._name, i.start or 0, i.stop - 1)
]
except AttributeError:
raise KeyError(f"Index should be an integer or a slice object. Was {i}")
if isinstance(i, slice):
return self._get_slice(i)
i = operator.index(i) # TypeError for non-integers, like list
v = self._source.lindex(self._name, i)
if v is None: # redis answers nil for an out of range index
raise IndexError("RedisList index out of range")
return self._decode(v)

def _get_slice(self, sl):
start, stop, step = sl.indices(len(self))
if step == 1:
if stop <= start:
return []
values = self._source.lrange(self._name, start, stop - 1)
else:
values = self._source.lrange(self._name, 0, -1)[sl]
return [self._decode(v) for v in values]

def __iter__(self):
# TODO: Find a more efficient way to do this.
# In batches perhaps? But what's the optimal batch size?
# async fetching of the next value/batch while yielding the one that's already fetched?
for i in range(len(self)):
yield self[i]
# One snapshot per batch (rather than one round trip per element)
n = max(1, int(self.iter_batch_size))
start = 0
while True:
batch = self._source.lrange(self._name, start, start + n - 1)
for v in batch:
yield self._decode(v)
if len(batch) < n:
return
start += n

def __setitem__(self, i, v):
return self._source.lset(self._name, i, self._encode(v))

def __delitem__(self, i): # TODO: Implement
raise NotImplementedError(
"Don't know how to (efficiently/directly) delete a single middle item."
)
if isinstance(i, slice):
raise NotImplementedError("Slice assignment is not supported")
i = operator.index(i)
try:
self._source.lset(self._name, i, self._encode(v))
except ResponseError as e: # redis: 'no such key' / 'index out of range'
raise IndexError("RedisList assignment index out of range") from e

def __delitem__(self, i):
"""Delete the element at position ``i`` (atomically).

LREM removes by *value*, so the element is first overwritten with a unique
sentinel, then that sentinel is removed, in one MULTI/EXEC transaction
(correct even when the list holds duplicate values).
"""
if isinstance(i, slice):
raise NotImplementedError("Slice deletion is not supported")
i = operator.index(i)
sentinel = _new_sentinel()
try:
with self._source.pipeline(transaction=True) as pipe:
pipe.lset(self._name, i, sentinel)
pipe.lrem(self._name, 1, sentinel)
pipe.execute()
except ResponseError as e:
raise IndexError("RedisList assignment index out of range") from e

def append(self, v):
# rpush (not rpushx): appending to a not-yet-existing list creates it,
Expand All @@ -277,8 +315,37 @@ def extend(self, iterable):
if values: # rpush with no values is a redis error
self._source.rpush(self._name, *values)

def insert(self, i, v): # TODO: Implement
raise NotImplementedError("Might have to do with self._source.linsert")
def clear(self):
self._source.delete(self._name)

def insert(self, i, v):
"""Insert ``v`` before position ``i`` (``list.insert`` semantics: ``i`` is clamped).

LINSERT is relative to a *value* (wrong with duplicates), so the element
currently at the target position is temporarily replaced by a unique
sentinel, ``v`` is inserted before it, and the original is put back. This
runs in a WATCH/MULTI transaction (retried if the list changes meanwhile).
"""
i = operator.index(i)
value = self._encode(v)
name = self._name

def _insert(pipe):
n = pipe.llen(name)
idx = i + n if i < 0 else i
idx = min(max(idx, 0), n)
if idx == n:
pipe.multi()
pipe.rpush(name, value)
return
original = pipe.lindex(name, idx)
sentinel = _new_sentinel()
pipe.multi()
pipe.lset(name, idx, sentinel)
pipe.linsert(name, "BEFORE", sentinel, value)
pipe.lset(name, idx + 1, original)

self._source.transaction(_insert, name)


# TODO: Consider if should move to stores
Expand Down Expand Up @@ -324,10 +391,19 @@ def __getitem__(self, k):

class RedisPersister(RedisCollection, RedisBytesPersister):
def __setitem__(self, k, v):
if isinstance(v, (str, bytes)):
return super().__setitem__(k, v)
elif isinstance(v, Iterable):
if k in self:
del self[k]
if isinstance(v, Iterable) and not isinstance(v, (str, bytes)):
encode = self._list_codec("list_encode") or _identity
return self._source.rpush(k, *map(encode, v))
values = [encode(x) for x in v] # also consumes one-shot iterables
if not values:
# Redis has no empty list (a list key with no elements doesn't exist).
# Refuse *before* touching the existing value, rather than lose it.
raise ValueError(
f"Can't store an empty list at {k!r}: Redis has no empty lists. "
"Use `del store[k]` to remove the key."
)
with self._source.pipeline(transaction=True) as pipe: # atomic replace
pipe.delete(k)
pipe.rpush(k, *values)
pipe.execute()
else: # str, bytes, numbers...: stored as a redis string
return super().__setitem__(k, v)
146 changes: 146 additions & 0 deletions tests/test_wrapping_fakeredis.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,3 +172,149 @@ def test_unsupported_redis_type_raises_typeerror_not_none():
assert "h" in s
with pytest.raises(TypeError, match="hash"):
s["h"]


# -- RedisList: MutableSequence contract (redisdol#6, #7) ----------------------------


def _rl(items=(), **kw):
l = RedisList(FakeRedis(), "l", **kw)
l.extend(items)
return l


def _dec(l):
return [bytes(x).decode() for x in l]


@pytest.mark.parametrize(
"idx",
[(None, None), (1, 0), (0, 0), (1, 3), (-2, None), (None, -1), (2, 99), (5, 9), (3, 1)],
)
def test_redislist_slices_match_python_list(idx):
ref = list("abcd")
l = _rl(ref)
sl = slice(*idx)
assert _dec(l[sl]) == ref[sl]


def test_redislist_stepped_slices_match_python_list():
ref = list("abcdef")
l = _rl(ref)
for sl in (slice(None, None, 2), slice(None, None, -1), slice(1, 5, 2)):
assert _dec(l[sl]) == ref[sl]


def test_redislist_index_errors():
l = _rl("ab")
assert l[-1] == b"b"
for bad in (2, 99, -3):
with pytest.raises(IndexError):
l[bad]
with pytest.raises(IndexError):
l[bad] = "x"
with pytest.raises(IndexError):
del l[bad]
with pytest.raises(IndexError):
RedisList(FakeRedis(), "nokey")[0]
assert _dec(l) == ["a", "b"] # failed ops left it untouched


def test_redislist_index_and_contains_terminate():
l = _rl("abc")
assert l.index(b"c") == 2
with pytest.raises(ValueError):
l.index(b"absent")
assert b"a" in l and b"z" not in l


def test_redislist_delitem_with_duplicates():
l = _rl("abab")
del l[2]
assert _dec(l) == ["a", "b", "b"]
del l[-1]
assert _dec(l) == ["a", "b"]
del l[0]
assert _dec(l) == ["b"]
del l[0]
assert len(l) == 0


def test_redislist_insert_matches_python_list_including_duplicates():
import random

rnd = random.Random(0)
for _ in range(50):
ref = [rnd.choice("ab") for _ in range(rnd.randrange(0, 6))]
l = _rl(ref)
i = rnd.randrange(-8, 9)
ref.insert(i, "X")
l.insert(i, "X")
assert _dec(l) == ref, (i,)


def test_redislist_insert_into_missing_list_and_codec():
l = RedisList(FakeRedis(), "new", encode=str.upper, decode=bytes.decode)
l.insert(0, "a")
l.insert(0, "b")
l.insert(1, "c")
assert list(l) == ["B", "C", "A"]


def test_redislist_mutablesequence_mixins():
l = _rl("abc")
assert l.pop() == b"c"
l.remove(b"a")
assert _dec(l) == ["b"]
l += [b"x", b"y"]
l.reverse()
assert _dec(l) == ["y", "x", "b"]
l.clear()
assert len(l) == 0 and not l._source.exists("l")


def test_redislist_iter_batches(monkeypatch):
l = _rl([str(i) for i in range(25)])
monkeypatch.setattr(RedisList, "iter_batch_size", 10)
assert _dec(l) == [str(i) for i in range(25)]
assert len(list(_rl([str(i) for i in range(20)]))) == 20


def test_redislist_slice_assign_and_delete_unsupported():
l = _rl("abc")
with pytest.raises(NotImplementedError):
l[0:1] = ["x"]
with pytest.raises(NotImplementedError):
del l[0:1]


def test_store_assign_empty_list_keeps_existing_value():
s = _store()
s["k"] = ["a", "b"]
with pytest.raises(ValueError):
s["k"] = []
assert list(s["k"]) == [b"a", b"b"]


def test_store_assign_replaces_list_and_accepts_generators_and_numbers():
s = _store()
s["k"] = ["a", "b"]
s["k"] = (x for x in "xyz")
assert list(s["k"]) == [b"x", b"y", b"z"]
s["n"] = 42
assert s["n"] == b"42"


def test_redislist_non_integer_index_is_typeerror(monkeypatch):
l = _rl("ab")
for bad in ("x", 1.5, None):
with pytest.raises(TypeError):
l[bad]
with pytest.raises(TypeError):
l[bad] = "v"
with pytest.raises(TypeError):
del l[bad]
with pytest.raises(TypeError):
l.insert(bad, "v")
monkeypatch.setattr(RedisList, "iter_batch_size", 0) # degenerate: still terminates
assert _dec(l) == ["a", "b"]
Loading