__all__ = ('DynArray',)
import collections.abc
import operator
import typing
from ._internal.desc_base_types import _u32_desc
from .core import (
ComplexCopyAction,
Slot,
SpecialTypeDesc,
TypeDesc,
_WithStorageSlotAndTD,
actions_apply_copy,
)
[docs]
class DynArray[T](_WithStorageSlotAndTD, collections.abc.MutableSequence[T]):
"""
Represents exponentially growing array (:py:class:`list` in python terms) that can be persisted on the blockchain
"""
_item_desc: TypeDesc
__slots__ = ('_item_desc', '_off', '_storage_slot')
[docs]
def __init__(self):
"""
This class can't be created with ``DynArray()``
:raises TypeError: always
"""
raise TypeError("this class can't be instantiated by user")
[docs]
def __len__(self) -> int:
return _u32_desc.get(self._storage_slot, self._off)
def _map_index(self, idx: int) -> int:
le = len(self)
if idx < 0:
idx += le
if idx < 0 or idx >= le:
raise IndexError(f'index out of range {idx} not in 0..<{le}')
return idx
@typing.overload
def __getitem__(self, idx: int) -> T: ...
@typing.overload
def __getitem__(self, idx: slice) -> list[T]: ...
[docs]
def __getitem__(self, idx: int | slice) -> T | list[T]:
"""
Get element by index or sublist by slice.
:param idx: integer index or slice
:returns: single element for int index, list of elements for slice
:raises IndexError: when integer index is out of range
"""
if isinstance(idx, int):
idx = self._map_index(idx)
items_at = self._storage_slot.indirect(self._off)
return self._item_desc.get(items_at, idx * self._item_desc.size)
else:
start, stop, step = idx.indices(len(self))
ret = []
step_sign = 1 if step >= 0 else -1
while start * step_sign < stop * step_sign:
ret.append(self[start])
start += step
return ret
@typing.overload
def __setitem__(self, idx: typing.SupportsIndex, val: T) -> None: ...
@typing.overload
def __setitem__(self, idx: slice, val: collections.abc.Iterable[T]) -> None: ...
[docs]
def __setitem__(
self, idx: typing.SupportsIndex | slice, val: T | collections.abc.Iterable[T]
) -> None:
"""
Set element by index or replace a range by slice.
:param idx: integer index or slice
:param val: value or sequence of values to assign
:raises IndexError: when integer index is out of range
If assigning an element or one of several slice elements fails, earlier
writes made by this operation remain visible. A failed extending slice
assignment changes the length only after all new elements are written.
"""
if not isinstance(idx, slice):
idx = self._map_index(idx.__index__())
items_at = self._storage_slot.indirect(self._off)
self._item_desc.set(items_at, idx * self._item_desc.size, val)
return
else:
start, stop, step = self._slice_to_idx(idx)
# materialized: the algorithm below needs `len` and reversal
new_val = list(typing.cast(collections.abc.Iterable[T], val))
left_in_new = len(new_val)
if isinstance(idx.step, int) and idx.step < 0:
new_val.reverse()
left_in_range = (stop - start) // step
new_it = iter(new_val)
# just reassign existing values
common_values_cnt = min(left_in_new, left_in_range)
for i in range(common_values_cnt):
self[start + i * step] = next(new_it)
start += common_values_cnt
left_in_range -= common_values_cnt
left_in_new -= common_values_cnt
# if we have other values we must remove them
if left_in_range > 0:
del self[start:stop:step]
# if we have some unassigned we must insert it here
elif left_in_new > 0:
# move current to the right
items_at = self._storage_slot.indirect(self._off)
for i in range(len(self) - 1, start - 1, -1):
self._item_desc.set(
items_at, (i + left_in_new) * self._item_desc.size, self[i]
)
for i in range(left_in_new):
self._item_desc.set(
items_at, (start + i) * self._item_desc.size, next(new_it)
)
_u32_desc.set(self._storage_slot, self._off, len(self) + left_in_new)
def _slice_to_idx(self, s: slice) -> tuple[int, int, int]:
start, stop, step = s.indices(len(self))
if step < 0:
step *= -1
start, stop = stop, start
# stop += (step - (stop - start) % step) % step
start = stop - (stop - start - 1) // step * step
stop += 1
return start, stop, step
@typing.overload
def __delitem__(self, idx: int) -> None: ...
@typing.overload
def __delitem__(self, idx: slice) -> None: ...
[docs]
def __delitem__(self, idx: int | slice) -> None:
"""
Delete element by index or range by slice.
:param idx: integer index or slice
:raises IndexError: when integer index is out of range
Elements are shifted before the length is reduced. If shifting fails,
the original length and any shifts already completed remain visible.
"""
if isinstance(idx, int):
start = self._map_index(idx)
stop = start + 1
step = 1
else:
start, stop, step = self._slice_to_idx(idx)
if stop <= start:
return
next_deletion = start
insert_idx = start
for i in range(start, len(self)):
if i == next_deletion:
next_deletion = i + step
if next_deletion >= stop:
next_deletion = -1
continue
self[insert_idx] = self[i]
insert_idx += 1
_u32_desc.set(self._storage_slot, self._off, insert_idx)
[docs]
def assign(self, arr: typing.Sequence[T], /) -> typing.Self:
"""
Same as ``self[:] = arr`` but more efficient
.. admonition:: Exception safety
:class: note
On error list becomes empty
"""
_u32_desc.set(self._storage_slot, self._off, 0)
for idx in range(len(arr)):
items_at = self._storage_slot.indirect(self._off)
self._item_desc.set(items_at, idx * self._item_desc.size, arr[idx])
_u32_desc.set(self._storage_slot, self._off, len(arr))
return self
[docs]
def insert(self, index: typing.SupportsIndex, value: T, /) -> None:
"""
Insert value before the given index.
:param index: position to insert at
:param value: value to insert
Like :py:meth:`list.insert`, negative indices are normalized and indices
outside the array are clamped to either end. The length is increased
before elements are shifted, so a failed write leaves the increased
length and any completed shifts visible.
"""
index = operator.index(index)
old_len = len(self)
if index < 0:
index = max(0, index + old_len)
else:
index = min(index, old_len)
_u32_desc.set(self._storage_slot, self._off, old_len + 1)
for i in range(old_len, index, -1):
self[i] = self[i - 1]
self[index] = value
[docs]
def __iter__(self) -> typing.Any:
for i in range(len(self)):
yield self[i]
[docs]
def append(self, value: T, /) -> None:
"""
Append value to the end of the array.
:param value: value to append
The length is increased before the value is written. If writing the
value fails, the new element remains visible with whatever data its
storage previously contained, or its zero-initialized value.
"""
le = len(self)
_u32_desc.set(self._storage_slot, self._off, le + 1)
items_at = self._storage_slot.indirect(self._off)
return self._item_desc.set(items_at, le * self._item_desc.size, value)
[docs]
def append_new_get(self) -> T:
"""
Grow the array by one and return a reference to the new (uninitialized) element.
:returns: reference to the newly appended element
The new element is not initialized by this method. It exposes the value
already present at its storage location, which is zero-initialized if the
location has never been written.
"""
le = len(self)
_u32_desc.set(self._storage_slot, self._off, le + 1)
items_at = self._storage_slot.indirect(self._off)
return self._item_desc.get(items_at, le * self._item_desc.size)
[docs]
def pop(self, index: typing.SupportsIndex = -1, /) -> T:
"""
Remove and return an element.
:param index: element to remove (default last)
:raises IndexError: when the array is empty or index is out of range
Storage-backed compound values are returned as views, not detached
Python objects. Removing a non-last element shifts another element into
the returned view's location; reusing the removed last slot can likewise
change a previously returned view.
"""
index = self._map_index(operator.index(index))
ret = self[index]
del self[index]
return ret
[docs]
def __repr__(self) -> str:
ret: list[str] = []
ret.append('[')
comma = False
for x in self:
if comma:
ret.append(',')
comma = True
ret.append(repr(x))
ret.append(']')
return ''.join(ret)
[docs]
def clear(self) -> None:
"""
Remove all elements from the array.
Payload bytes remain in storage; later growth can expose them again.
"""
_u32_desc.set(self._storage_slot, self._off, 0)
class _DynArrayDesc(SpecialTypeDesc, ComplexCopyAction):
__slots__ = ('item_desc', 'view_ctor')
def __init__(self, item_desc: TypeDesc):
SpecialTypeDesc.__init__(self, item_desc, lambda: DynArray.__new__(DynArray))
TypeDesc.__init__(self, _u32_desc.size, [self])
def copy(self, frm: Slot, frm_off: int, to: Slot, to_off: int) -> int:
le = _u32_desc.get(frm, frm_off)
_u32_desc.set(to, to_off, le)
cop = self.item_desc.copy_actions
to_indirect = to.indirect(to_off)
frm_indirect = frm.indirect(frm_off)
if len(cop) == 1 and isinstance(cop[0], int):
to_indirect.write(0, frm_indirect.read(0, cop[0] * le))
else:
cum_off = 0
for _i in range(le):
cum_off += actions_apply_copy(cop, to_indirect, cum_off, frm_indirect, cum_off)
return _u32_desc.size
def set(self, slot: Slot, off: int, val: DynArray | collections.abc.Sequence) -> None:
if isinstance(val, DynArray):
if val._item_desc is not self.item_desc:
raise TypeError('incompatible vector type')
self.copy(val._storage_slot, val._off, slot, off)
return
_u32_desc.set(slot, off, len(val))
indirect_slot = slot.indirect(off)
for i in range(len(val)):
self.item_desc.set(indirect_slot, i * self.item_desc.size, val[i])
return