Source code for eventsourcing.domain.model.array

from abc import abstractmethod
from math import ceil, log
from typing import Any
from uuid import uuid5

from eventsourcing.domain.model.decorators import retry
from eventsourcing.domain.model.entity import (
from import publish
from eventsourcing.exceptions import ArrayIndexError, ConcurrencyError
from eventsourcing.infrastructure.base import AbstractEntityRepository

[docs]class ItemAssigned(TimestampedVersionedEntity.Event): """Occurs when an item is set at a position in an array."""
[docs] def __init__(self, item, index, **kwargs): kwargs["item"] = item super(ItemAssigned, self).__init__(originator_version=index, **kwargs)
@property def item(self): return self.__dict__["item"] @property def index(self): return self.originator_version
class Array(object): def __init__(self, array_id, repo: "AbstractArrayRepository"): = array_id self.repo = repo @retry(ConcurrencyError, max_attempts=50, wait=0.01) def append(self, item): """Sets item in next position after the last item.""" self.__setitem__(self.get_next_position(), item) def __setitem__(self, index, item): """ Sets item in array, at given index. Won't overrun the end of the array, because the position is fixed to be less than base_size. """ size = self.repo.array_size if size and index >= size: raise ArrayIndexError("Index is {}, but size is {}".format(index, size)) event = ItemAssigned(, index=index, item=item) publish([event]) def __getitem__(self, item): """ Returns item at index, or items in slice. """ assert isinstance(item, (int, slice)) array_size = self.repo.array_size if isinstance(item, int): if item < 0: index = array_size + item if index < 0: raise ArrayIndexError("Array index out of range: {}".format(item)) else: index = item event = self.get_item_assigned(index) assert isinstance(event, ItemAssigned) return event.item elif isinstance(item, slice): assert item.step in (None, 1), "Slice stepping not supported, yet" if item.start is None: start_index = 0 elif item.start < 0: start_index = max(array_size + item.start, 0) else: start_index = item.start if not isinstance(item.stop, int): stop_index = array_size elif item.stop < 0: stop_index = array_size + item.stop else: stop_index = item.stop if stop_index is not None: if stop_index <= 0: return [] # Avoid passing massive integers into the database. if array_size is not None: stop_index = min(stop_index, self.repo.array_size) items_assigned = self.get_items_assigned( stop_index=stop_index, start_index=start_index ) items_dict = {str(i.originator_version): i.item for i in items_assigned} items = [items_dict.get(str(i)) for i in range(start_index, stop_index)] return items def __len__(self): """ Returns length of array. """ return self.repo.array_size def get_next_position(self): """ Returns next 'position' in 0-based sequence. :return: Non-negative integer. :rtype: int """ return self.get_last_item_and_next_position()[1] def get_last_item_and_next_position(self): # Get last assigned item. items_assigned = list(self.get_items_assigned(limit=1, is_ascending=False)) if len(items_assigned) == 0: return None, 0 item_assigned = items_assigned[0] # Add 1 to the index to get the next position (in zero-based sequence). next_postion = item_assigned.index + 1 return item_assigned.item, next_postion def get_items_assigned( self, start_index=None, stop_index=None, limit=None, is_ascending=True ): return self.repo.event_store.iter_events(, gte=start_index, lt=stop_index, limit=limit, is_ascending=is_ascending, ) def get_item_assigned(self, index): try: item_assigned = self.repo.event_store.get_event(, position=index ) except IndexError as e: raise ArrayIndexError(e) else: return item_assigned def __eq__(self, other): return isinstance(other, type(self)) and == def __ne__(self, other): return not self.__eq__(other)
[docs]class BigArray(Array): """ A virtual array holding items in indexed positions, across a number of Array instances. Getting and setting items at index position is supported. Slices are supported, and operate across the underlying arrays. Appending is also supported. BigArray is designed to overcome the concern of needing a single large sequence that may not be suitably stored in any single partiton. In simple terms, if events of an aggregate can fit in a partition, we can use the same size partition to make a tree of arrays that will certainly be capable of sequencing all the events of the application in a single stream. With normal size base arrays, enterprise applications can expect read and write time to be approximately constant with respect to the number of items in the array. The array is composed of a tree of arrays, which gives the capacity equal to the size of each array to the power of the size of each array. If the arrays are limited to be about the maximum size of an aggregate event stream (a large number but not too many that would cause there to be too much data in any one partition, let's say 1000s to be safe) then it would be possible to fit such a large number of aggregates in the corresponding BigArray, that we can be confident it would be full. Write access time in the worst case, and the time to identify the index of the last item in the big array, is proportional to the log of the highest assigned index to base the underlying array size. Write time on average, and read time given an index, is constant with respect to the number of items in a BigArray. Items can be appended in log time in a single thread. However, the time between reading the current last index and claiming the next position leads to contention and retries when there are lots of threads of execution all attempting to append items, which inherently limits throughput. Todo: Not possible in Cassandra, but maybe do it in a transaction in SQLAlchemy? An alternative to reading the last item before writing the next is to use an integer sequence generator to generate a stream of integers. Items can be assigned to index positions in a big array, according to the integers that are issued. Throughput will then be much better, and will be limited only by the rate at which the database can have events written to it (unless the number generator is quite slow). An external integer sequence generator, such as Redis' INCR command, or an auto-incrementing database column, may constitute a single point of failure. """
[docs] def __init__(self, array_id, repo): super(BigArray, self).__init__(array_id=array_id, repo=repo)
[docs] def get_last_array(self): """ Returns last array in compound. :rtype: CompoundSequenceReader """ # Get the root array (might not have been registered). root = self.repo[] # Get length and last item in the root array. apex_id, apex_height = root.get_last_item_and_next_position() # Bail if there isn't anything yet. if apex_id is None: return None, None # Get the current apex array. apex = self.repo[apex_id] assert isinstance(apex, Array) # Descend until hitting the bottom. array = apex array_i = 0 height = apex_height while height > 1: height -= 1 array_id, width = array.get_last_item_and_next_position() assert width > 0 offset = width - 1 array_i += offset * self.repo.array_size ** height array = self.repo[array_id] return array, array_i
def get_last_item_and_next_position(self): sequence, i = self.get_last_array() if sequence is None: return None, 0 item, n = sequence.get_last_item_and_next_position() return item, i + n
[docs] def __getitem__(self, item): assert isinstance(item, (int, slice)) # Calculate the i and j of the containing base sequence. if isinstance(item, int): return self.get_item(item) elif isinstance(item, slice): assert item.step in (None, 1), "Slice step must be 1: {}".format( str(item.step) ) return self.get_slice(item.start, item.stop)
def get_item(self, position): if position < 0: last, length = self.get_last_item_and_next_position() if position == -1: return last position = position + length size = self.repo.array_size n = position // size i = n * size j = i + size offset = position - i array_id = self.create_array_id(i, j) base_array = self.repo[array_id] return base_array[offset] def get_slice(self, start, stop): array_len = self.repo.array_size ** self.repo.array_size if start is None: start = 0 elif start < 0: start = max(array_len + start, 0) if stop is None: stop = array_len elif stop < 0: stop = array_len + stop stop = min(stop, array_len) size = self.repo.array_size while start < stop: n = start // size i = n * size j = i + size substart = start - i substop = stop - i array_id = self.create_array_id(i, j) sequence = self.repo[array_id] for item in sequence[substart:substop]: yield item start = j
[docs] def __setitem__(self, position, item): # Calculate start and stop position # of the containing base array. size = self.repo.array_size start = (position // size) * size stop = start + size # Set the item in the base array. array_id = self.create_array_id(start, stop) array = self.repo[array_id] index = position - start array[index] = item # Calculate the height of the apex of # the compound containing given position # (zero-based index in big array). required_height = self.calc_required_height(position, size) # Set array IDs in containing arrays, # up to the required height. height = 1 while height < required_height: # Set ID of array in its parent array. start, stop, height, index_of_child = self.calc_parent(start, stop, height) child_id = array_id array_id = self.create_array_id(start, stop) array = self.repo[array_id] try: array[index_of_child] = child_id except ConcurrencyError: return # Set the apex to the root array. assert array_id root = self.repo[] try: root[required_height - 1] = array_id except ConcurrencyError: return
[docs] def __len__(self): """ Returns length of array. """ return self.repo.array_size ** self.repo.array_size
def calc_required_height(self, n, size): if size <= 0: raise ValueError("Size must be greater than 0") capacity = size ** size if n + 1 > capacity: raise ArrayIndexError("Contents can't be greater than capacity") if n == 0 or n == 1: return 1 min_capacity = max(int(size), int(n + 1)) required_height = log(min_capacity, int(size)) # Quash numerical error before calling ceil. # - this is occasionally required, for example # with base size of 1000 numerical error # makes the required height one greater # than the correct value, which causes # an index error when assigning apex # ID to the root sequence. required_height = round(required_height, 10) return int(ceil(required_height))
[docs] def calc_parent(self, i, j, h): """ Returns get_big_array and end of span of parent sequence that contains given child. """ N = self.repo.array_size c_i = i c_j = j c_h = h # Calculate the number of the sequence in its row (sequences # with same height), from left to right, starting from 0. c_n = c_i // (N ** c_h) p_n = c_n // N # Position of the child ID in the parent array. p_p = c_n % N # Parent height is child height plus one. p_h = c_h + 1 # Span of sequences in parent row is max size N, to the power of the height. span = N ** p_h # Calculate parent i and j. p_i = p_n * span p_j = p_i + span # Check the parent i,j bounds the child i,j, ie child span is contained by # parent span. assert p_i <= c_i, "i greater on parent than child: {}".format((p_i, p_j)) assert p_j >= c_j, "j less on parent than child: {}".format((p_i, p_j)) # Return parent i, j, h, p. return p_i, p_j, p_h, p_p
def create_array_id(self, i, j): return uuid5(, str((i, j)))
[docs]class AbstractArrayRepository(AbstractEntityRepository[Any, ItemAssigned]): """ Repository for sequence objects. """
[docs] def __init__(self, array_size=10000, *args, **kwargs): super(AbstractArrayRepository, self).__init__(*args, **kwargs) self.array_size = array_size
[docs] def __getitem__(self, entity_id) -> Any: """ Returns sequence for given ID. """ return Array(array_id=entity_id, repo=self)
[docs]class AbstractBigArrayRepository(AbstractEntityRepository): """ Repository for compound sequence objects. """ @property @abstractmethod def subrepo(self): """Sub-sequence repository."""
[docs] def __getitem__(self, entity_id) -> Any: """ Returns sequence for given ID. """ return BigArray(array_id=entity_id, repo=self.subrepo)