Source code for rlp.lazy

from collections import Iterable, Sequence
from .codec import consume_length_prefix, consume_payload
from .exceptions import DecodingError
from .utils import Atomic


[docs]def decode_lazy(rlp, sedes=None, **sedes_kwargs): """Decode an RLP encoded object in a lazy fashion. If the encoded object is a bytestring, this function acts similar to :func:`rlp.decode`. If it is a list however, a :class:`LazyList` is returned instead. This object will decode the string lazily, avoiding both horizontal and vertical traversing as much as possible. The way `sedes` is applied depends on the decoded object: If it is a string `sedes` deserializes it as a whole; if it is a list, each element is deserialized individually. In both cases, `sedes_kwargs` are passed on. Note that, if a deserializer is used, only "horizontal" but not "vertical lazyness" can be preserved. :param rlp: the RLP string to decode :param sedes: an object implementing a method ``deserialize(code)`` which is used as described above, or ``None`` if no deserialization should be performed :param \*\*sedes_kwargs: additional keyword arguments that will be passed to the deserializers :returns: either the already decoded and deserialized object (if encoded as a string) or an instance of :class:`rlp.LazyList` """ item, end = consume_item_lazy(rlp, 0) if end != len(rlp): raise DecodingError('RLP length prefix announced wrong length', rlp) if isinstance(item, LazyList): item.sedes = sedes item.sedes_kwargs = sedes_kwargs return item elif sedes: return sedes.deserialize(item, **sedes_kwargs) else: return item
def consume_item_lazy(rlp, start): """Read an item from an RLP string lazily. If the length prefix announces a string, the string is read; if it announces a list, a :class:`LazyList` is created. :param rlp: the rlp string to read from :param start: the position at which to start reading :returns: a tuple ``(item, end)`` where ``item`` is the read string or a :class:`LazyList` and ``end`` is the position of the first unprocessed byte. """ t, l, s = consume_length_prefix(rlp, start) if t == str: #item, _ = consume_payload(rlp, s, str, l), s + l return consume_payload(rlp, s, str, l) else: assert t == list return LazyList(rlp, s, s + l), s + l
[docs]class LazyList(Sequence): """A RLP encoded list which decodes itself when necessary. Both indexing with positive indices and iterating are supported. Getting the length with :func:`len` is possible as well but requires full horizontal encoding. :param rlp: the rlp string in which the list is encoded :param start: the position of the first payload byte of the encoded list :param end: the position of the last payload byte of the encoded list :param sedes: a sedes object which deserializes each element of the list, or ``None`` for no deserialization :param \*\*sedes_kwargs: keyword arguments which will be passed on to the deserializer """ def __init__(self, rlp, start, end, sedes=None, **sedes_kwargs): self.rlp = rlp self.start = start self.end = end self.index = start self._elements = [] self._len = None self.sedes = sedes self.sedes_kwargs = sedes_kwargs def next(self): if self.index == self.end: self._len = len(self._elements) raise StopIteration assert self.index < self.end item, end = consume_item_lazy(self.rlp, self.index) self.index = end if self.sedes: item = self.sedes.deserialize(item, **self.sedes_kwargs) self._elements.append(item) return item def __getitem__(self, i): try: while len(self._elements) <= i: self.next() except StopIteration: assert self.index == self.end raise IndexError('Index %d out of range' % i) return self._elements[i] def __len__(self): if not self._len: try: while True: self.next() except StopIteration: self._len = len(self._elements) return self._len
def peek(rlp, index, sedes=None): """Get a specific element from an rlp encoded nested list. This function uses :func:`rlp.decode_lazy` and, thus, decodes only the necessary parts of the string. Usage example:: >>> rlpdata = rlp.encode([1, 2, [3, [4, 5]]]) >>> rlp.peek(rlpdata, 0, rlp.sedes.big_endian_int) 1 >>> rlp.peek(rlpdata, [2, 0], rlp.sedes.big_endian_int) 3 :param rlp: the rlp string :param index: the index of the element to peek at (can be a list for nested data) :param sedes: a sedes used to deserialize the peeked at object, or `None` if no deserialization should be performed :raises: :exc:`IndexError` if `index` is invalid (out of range or too many levels) """ ll = decode_lazy(rlp) if not isinstance(index, Iterable): index = [index] for i in index: if isinstance(ll, Atomic): raise IndexError('Too many indices given') ll = ll[i] if sedes: return sedes.deserialize(ll) else: return ll