-
Notifications
You must be signed in to change notification settings - Fork 3
/
listish.py
128 lines (102 loc) · 3.71 KB
/
listish.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
"""
A collection of lazy, repeatably consumable list- & tuple-like objects which
wrap any iterator.
"""
from collections import Sequence, MutableSequence
from sys import version_info
if version_info < (3, 0): # pragma: no cover
range = xrange # pylint: disable=redefined-builtin,invalid-name
__version__ = '1.0'
class MustExhaustException(RuntimeError):
"""
Raised when a slice cannot be converted into a list of indexes as it
requires knowledge of the length of the iterable
"""
class Tupleish(Sequence):
"""
A lazily evaluated tuple-like object.
"""
def __init__(self, iterable):
# iter(iterator) returns the iterator
self._iterator = iter(iterable)
self._datastore = []
def _consume_next(self):
value = next(self._iterator)
self._datastore.append(value)
return value
def __iter__(self):
for value in self._datastore:
yield value
while True:
# _consume_next will already raise StopIteration when it is
# exhausted
yield self._consume_next()
def __getitem__(self, index):
if isinstance(index, slice):
start, stop, step = self._get_indices(index)
min_slice = min(start, stop)
# Add two to the max to ensure we get the last value (+1) plus the
# fact that range() is exclusive on its stop value and requires an
# additional +1 to cover the expected inclusive stop
max_slice = max(start, stop) + 2
maximum_possible_slice = range(min_slice, max_slice)
list(self._get_items_in_bounds(maximum_possible_slice))
return list(self._get_items_in_bounds(range(start, stop, step)))
try:
while len(self._datastore) < index + 1:
self._consume_next()
return self._datastore[index]
except StopIteration:
raise IndexError("index out of range")
def _get_items_in_bounds(self, indexes):
for i in indexes:
try:
yield self[i]
except IndexError:
return
def _get_indices(self, slice_):
try:
unbounded_end = slice_.start is not None and slice_.stop is None
unbounded_step = (
slice_.start is None and
slice_.stop is None and
slice_.step is not None
)
negative_index = (
(slice_.start is not None and slice_.start < 0) or
(slice_.stop is not None and slice_.stop < 0)
)
if unbounded_end or unbounded_step or negative_index:
raise MustExhaustException()
length = max(slice_.stop, slice_.start) + 1
except MustExhaustException:
length = len(self)
return slice_.indices(length)
def __len__(self):
self._datastore += list(self._iterator)
return len(self._datastore)
class Listish(Tupleish, MutableSequence):
"""
A lazily evaluated list-like object.
"""
def __setitem__(self, index, value):
try:
self[index]
except IndexError:
raise IndexError("assignment index out of range")
self._datastore[index] = value
def __delitem__(self, index):
try:
self[index]
except IndexError:
raise IndexError("assignment index out of range")
del self._datastore[index]
def insert(self, index, value):
if index == 0:
self._datastore = [value] + self._datastore
else:
try:
self[index - 1]
except IndexError:
pass
self._datastore.insert(index, value)