satella.coding.structures package¶
Subpackages¶
- satella.coding.structures.dictionaries package
- Submodules
- satella.coding.structures.dictionaries.cache_dict module
- satella.coding.structures.dictionaries.counting module
- satella.coding.structures.dictionaries.default module
- satella.coding.structures.dictionaries.dict_object module
- satella.coding.structures.dictionaries.expiring module
- satella.coding.structures.dictionaries.objects module
- satella.coding.structures.dictionaries.writeback_cache module
- Module contents
- satella.coding.structures.heaps package
- satella.coding.structures.mixins package
Submodules¶
satella.coding.structures.hashable_objects module¶
- class satella.coding.structures.hashable_objects.HashableWrapper(object_to_wrap, wrap_operations=False)¶
Bases:
Proxy
A class that makes given objects hashable by their id, if the object is already not hashable.
Also overrides __eq__
Note that this class will return a proxy to the object, and not the object itself.
Use like:
>>> a = {1:2, 3:4} >>> a = HashableWrapper(a) >>> hash(a)
- Parameters:
object_to_wrap (T) –
wrap_operations (bool) –
satella.coding.structures.immutable module¶
- class satella.coding.structures.immutable.Immutable(*args, **kwargs)¶
Bases:
object
A mix-in to make your classes immutable.
You can assign normally using your constructor.
>>> class Test(Immutable): >>> def __init__(self): >>> self.attribute = 'value'
- class satella.coding.structures.immutable.ImmutableMetaType(name, bases, namespace, /, **kwargs)¶
Bases:
ABCMeta
- class satella.coding.structures.immutable.NotEqualToAnything¶
Bases:
object
A class that defines __eq__ and __ne__ to be always False and True respectively
When exploiting this class please make sure that it gets to be the first operand in the comparison, so that it’s __eq__ and __ne__ gets called!
- class satella.coding.structures.immutable.frozendict¶
Bases:
dict
A hashable dict with express forbid to change it’s values Both keys and values must be hashable in order for this dict to be hashable.
- update([E, ]**F) None. Update D from dict/iterable E and F. ¶
If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]
satella.coding.structures.lru module¶
- class satella.coding.structures.lru.LRU¶
Bases:
Generic
[T
]A class to track least recently used objects.
- add(item)¶
- Parameters:
item (T) –
- Return type:
None
- get_item_to_evict()¶
Return a least recently used object
- Return type:
T
- mark_as_used(item)¶
- Parameters:
item (T) –
- remove(item)¶
- Parameters:
item (T) –
- Return type:
None
satella.coding.structures.proxy module¶
- class satella.coding.structures.proxy.Proxy(object_to_wrap, wrap_operations=False)¶
Bases:
Generic
[T
]A base class for classes that try to emulate some other object.
They will intercept all calls and place them on the target object.
Note that in-place operations will return the Proxy itself, whereas simple addition will shed this proxy, returning object wrapped plus something.
Only __isinstance__ check is not overridden, due to Python limitations.
Please access this object in your descendant classes via
>>> getattr(self, '_Proxy__obj')
Please note that this class does not overload the descriptor protocol, not the pickle interface!
Note that this overloads __repr__, __str__ and __dir__, which may prove confusing. Handle this in your descendant classes.
If wrap_operations is set, the following code will be executed on the result:
>>> a = a.__add__(b) >>> return self.__class__(a)
Wrapped operations include all arithmetic operations and bitwise operations, except for divmod, but including concat, rounding, truncing and ceiling.
- Parameters:
object_to_wrap (T) – object to wrap
wrap_operations (bool) – whether results of operations returning something else should be also proxied.
- satella.coding.structures.proxy.wrap_operation(fun)¶
satella.coding.structures.push_iterable module¶
- class satella.coding.structures.push_iterable.PushIterable(iterable)¶
Bases:
Generic
[T
]An iterable that you can add elements to it’s head, and they will be popped before this iterable is consumed when a pop is called.
Example:
a = PushIterable([1, 2, 3]) assert a.pop() == 1 a.push(0) assert a.pop() == 0 assert a.pop() == 2 a.push(0) a.push(1) assert a.pop() == 1 assert a.pop() == 0 a.push_left(0) a.push_left(1) assert a.pop() == 0 assert a.pop() == 1 assert a.pop() == 3 assertRaises(StopIteration, a.pop)
- Parameters:
iterable (Iterable[T]) –
- pop()¶
Return next element from the stack :return: a next element
- Return type:
T
- push(item)¶
Push an item so that next pop will retrieve this item
- Parameters:
item (T) – item to push
- Return type:
None
- push_left(item)¶
Push an item so that last pop from internal buffer will retrieve this item
- Parameters:
item – item to push
satella.coding.structures.queues module¶
- class satella.coding.structures.queues.Subqueue¶
Bases:
Generic
[T
]Or a named queue is a collection of thread safe queues identified by name
- assert_queue(queue_name)¶
Assure that we have a queue with a particular name in the dictionary
- Parameters:
queue_name (str) –
- Return type:
None
- get(queue_name, block=True, timeout=None)¶
Same semantics at queue.Queue.get
- Parameters:
queue_name (str) –
- Return type:
T
- get_any()¶
Return any message with name of the queue.
This assumes block=False
- Returns:
a tuple of (queue_name, object)
- Raises:
queue.Empty()
- Return type:
Tuple[str, T]
- put(queue_name, obj)¶
Same semantics as queue.Queue.put
- Parameters:
queue_name (str) –
obj (T) –
- Return type:
None
- qsize()¶
Calculate the total of entries
- Return type:
int
satella.coding.structures.ranking module¶
- class satella.coding.structures.ranking.Ranking(items, key)¶
Bases:
Generic
[T
]A set of objects with them being ranked by their single property with the assumption that this property changes only when we want to.
Positions are counted from 0, where 0 has the least key value.
Essentially, this is a SortedList with the option to query at which position can be given element found.
Example usage:
>>> Entry = collections.namedtuple('Entry', ('key', )) >>> e1 = Entry(2) >>> e2 = Entry(3) >>> e3 = Entry(5) >>> ranking = Ranking([e1, e2, e3], lambda e: e.key) >>> assert ranking[0] == e1 # Get the first element >>> assert ranking[-1] == e3 # Get the last element >>> assert ranking.get_position_of(e1) == 0
- Parameters:
items (List[T]) –
key (Callable[[T], int]) –
- add(item)¶
Add a single element to the ranking and _recalculate it
- Parameters:
item (T) –
- Return type:
None
- calculate_ranking_for(item)¶
- Parameters:
item (T) –
- Return type:
int
- element_to_position: Dict[int, int]¶
- get_position_of(item)¶
Return the position in the ranking of element item
- Parameters:
item (T) – element to return the position for
- Returns:
position
- Raises:
ValueError – this element is not in the ranking
- Return type:
int
- get_sorted()¶
Return all the elements sorted from the least key value to the highest key value
- Return type:
Iterator[T]
- items¶
- key¶
- ranking: SortedList[T]¶
- remove(item)¶
Remove a single element from the ranking and _recalculate it
- Parameters:
item (T) –
- Return type:
None
satella.coding.structures.singleton module¶
- satella.coding.structures.singleton.Singleton(cls)¶
Make a singleton out of decorated class.
Usage:
>>> @Singleton >>> class MyClass: >>> ...
- satella.coding.structures.singleton.SingletonWithRegardsTo(num_args, weak_refs=False)¶
Make a memoized singletion depending on the arguments.
A dictionary is made (first N arguments => class instance) and such is returned. Please take care to ensure that a tuple made out of first num_args can be used as a dictionary key (ie. is both hashable and __eq__-able).
Usage:
>>> @SingletonWithRegardsTo(num_args=1) >>> class MyClass: >>> def __init__(self, device_id: str): >>> ... >>> a = MyClass('dev1') >>> b = MyClass('dev2') >>> c = MyClass('dev1') >>> assert a is c >>> assert b is not c
- Parameters:
num_args (int) – number of arguments to consume
weak_refs (bool) – if True, then singleton will be stored within a weak dictionary, so that it cleans up after itself when the values are gone.
Warning
If you set weak_refs to False and have a potentially unbounded number of argument values, you better watch out for the memory usage.
- satella.coding.structures.singleton.delete_singleton_for(x, *args)¶
Delete singleton for given arguments in a class decorated with SingletonWithRegardsTo
- Parameters:
x – class decorated with SingletonWithRegardsTo
args – arguments used in the constructor
- Return type:
None
- satella.coding.structures.singleton.get_instances_for_singleton(x)¶
Obtain a list of arguments for which singletons exists for given class decorated with SingletonWithRegardsTo
- Parameters:
x – a class decorated with SingletonWithRegardsTo
- Returns:
a list of arguments
- Return type:
List[Tuple]
satella.coding.structures.sorted_list module¶
- class satella.coding.structures.sorted_list.SliceableDeque(*args, **kwargs)¶
Bases:
MutableSequence
,Generic
[T
]A collections.deque that supports slicing.
Just note that it will return a p_gen upon being sliced!
- deque¶
- insert(i, item)¶
S.insert(index, value) – insert value before index
- Parameters:
i (int) –
item (T) –
- class satella.coding.structures.sorted_list.SortedList(items=(), key=<function SortedList.<lambda>>)¶
Bases:
Generic
[T
]An always-sorted sort of a set
It is assumed that keys of constituent elements don’t change.
list[0] will have the smallest element, and list[-1] the biggest.
Addition is O(n) worst-case, so is deletion.
- Parameters:
items (Iterable[T]) – items to construct the list with
key (Callable[[T], int]) – a callable[T]->int that builds the key of the sort
- add(other)¶
Add an element. Returns the index at which it was inserted.
- Parameters:
other (T) – element to insert
- Returns:
index that the entry is available now at
- Return type:
int
- extend(elements)¶
Adds multiple elements to this list
- Parameters:
elements (Iterable[T]) –
- index(other)¶
Return index at which given value has been placed
- Parameters:
other (T) –
- Return type:
int
- items: SliceableDeque[T]¶
- key: Callable[[T], int]¶
- keys: deque[int]¶
- pop()¶
Return the highest element, removing it from the list
- Return type:
T
- popleft()¶
Return the smallest element, removing it from the list
- Return type:
T
- remove(other)¶
Remove an element from the list
- Parameters:
other (T) – element to remove
- Raises:
ValueError – element not in list
- Return type:
None
satella.coding.structures.sparse_matrix module¶
- class satella.coding.structures.sparse_matrix.SparseMatrix(matrix_data=None)¶
Bases:
Generic
[T
]A matrix of infinite size, that supports assignments.
Set elements like this:
>>> sm[1, 2] = 5 >>> sm[:,1] = [5] >>> sm[1,:] = [5] >>> sm[:,:] = [[5]]
where the first argument is the number of the column, counted from 0, and the second one is the number of the row, also counted from 0
Note that custom slicing (ie. slices which are not :) will not be supported. Negative indices are supported.
Undefined elements are considered to be of value None.
Iterating over this matrix will yield it’s consecutive rows.
You can use the constructor in following way:
>>> sm = SparseMatrix([[1, 2], [3, 4]])
to construct a matrix that looks like
|1 2| |3 4|
- Parameters:
matrix_data (Optional[List[List[T]]]) –
- append_row(y)¶
Append a row to the bottom of the matrix
- Parameters:
y (Iterable[T]) – iterable with consequent columns
- Return type:
None
- clear()¶
Clear the contents of the sparse matrix
- Return type:
None
- property columns: int¶
Return the amount of columns
- delete_row(row_no)¶
Delete a row with specified number
- Parameters:
row_no (int) – number of the row to delete
- Return type:
None
- classmethod from_iterable(y)¶
Construct a sparse matrix given a row-first iterable. That iterable must return another iterable, that will yield values for given column.
- Parameters:
y (Iterable[Iterable[T]]) – an iterable describing the sparse matrix
- Returns:
a sparse matrix object
- get_neighbour_coordinates(col, row, include_diagonals=True)¶
Return an iterator of coordinates to points neighbouring given point. :param col: column :param row: row :param include_diagonals: whether to include points having only a single point in common :return: an iterable of coordinates of neighbouring points. An iterator of tuple
(col, row)
- Parameters:
col (int) –
row (int) –
include_diagonals (bool) –
- Return type:
Iterator[Tuple[int, int]]
- get_row(row_no)¶
Return a single row of provided number.
The returned array has the same length as .columns
- Parameters:
row_no (int) – row number, numbered from 0
- Return type:
List[T]
- known_column_count¶
- max()¶
Return maximum element.
None elements will be ignored.
- Return type:
T
- min()¶
Return minimum element.
None elements will be ignored.
- Return type:
T
- no_cols¶
- no_rows¶
- property rows: int¶
Return the amount of rows
- rows_dict¶
- shoot()¶
Insert an empty cell between current cells. So the matrix which looked like [[1, 2], [3, 4]] will now look like [[1, None, 2], [None, None, None], [3, None, 4]]
- Return type:
satella.coding.structures.syncable_droppable module¶
- class satella.coding.structures.syncable_droppable.DBStorage¶
Bases:
object
An abstract implementation of the storage class provided to
SyncableDroppable
This serves as satella’s hook into your database infrastructure.
- abstract delete(key)¶
Called by SyncableDroppable when there’s a need to remove target key
- Parameters:
key (K) – key to remove
- Return type:
None
- abstract iterate(starting_key)¶
Return an iterator iterating from provided starting key to the end of the values, as read from the database.
This may block for a while.
This iterator will be closed upon no longer being necessary.
- Parameters:
starting_key (Optional[K]) – starting key, included, or None for iterate from the start
- Returns:
an iterator from provided key (included) to the end of the range
- Return type:
Iterator[Tuple[K, V]]
- abstract on_change_start_entry(start_entry)¶
Called by SyncableDroppable when starting entry (earliest entry encountered both in the DB and is memory) is changed.
- Parameters:
start_entry (Optional[K]) – new value of start entry or None if there are no entries at all
- Return type:
None
- abstract on_change_stop_entry(stop_entry)¶
Called by SyncableDroppable when stopping entry (earliest entry encountered both in the DB and is memory) is changed.
- Parameters:
stop_entry (Optional[K]) – new value of stop entry or None if there are no entries at all
- Return type:
None
- abstract on_change_synced_up_to(synced_up_to)¶
Called by SyncableDroppable when synced up to (earliest timestamp synced) is changed.
- Parameters:
synced_up_to (Optional[K]) – new value of synced up to
- Return type:
None
- abstract put(key, value)¶
Put given value to storage at given key.
This may block for a while.
- Parameters:
key (K) – key to store
value (V) – value to store
- Return type:
None
- class satella.coding.structures.syncable_droppable.SyncableDroppable(db_storage, start_entry, stop_entry, synced_up_to, span_to_keep_in_memory, span_to_keep_in_db)¶
Bases:
RMonitor
,Generic
[K
,V
]A thread-safe class representing some single time series, which needs to be synchronized with some server, and may be too large to keep in memory. Moreover, after the sync we need to retain a part of the time series in memory for future requests. Only series past some timestamp may be deleted.
For brevity, this will refer to keys as timestamps. The keys must be __eq__able, comparable and subtractable.
A rule is that an item can never be both in memory and in the DB.
- Parameters:
db_storage (DBStorage) – a DBStorage implementation of your own provision, that serves as class’ interface with the database
start_entry (Optional[K]) – earliest timestamp stored or None if no data is stored
stop_entry (Optional[K]) – latest timestamp stored or None if no data is stored
synced_up_to (Optional[K]) – timestamp of the last entry synchronized or None if no data is stored
span_to_keep_in_memory (int) – key span to keep in memory. Entries earlier than difference of the latest key and this will be dropped from memory, onto the DB. Can’t be false.
span_to_keep_in_db (int) – key span to keep on disk. Entries earlier than difference of the latest key and this will be dropped from the DB. Can’t be false.
Note
Note that proper handling of maximum spans requires periodical calls to
cleanup()
You need to invoke this at your constructor You can also use it to release locks of other objects.
- cleanup()¶
Make sure that everything’s that in memory and the DB conforms to span_to_keep_in_db and span_to_keep_in_memory.
This may block for a while.
Warning
This bugs out for now. For now, a UserWarning will be raised and nothing will be done.
- Return type:
None
- cleanup_keep_in_db()¶
Clear up the database to conform to our span_to_keep_in_db
- Return type:
None
- cleanup_keep_in_memory()¶
Eject values from memory that should reside in the DB onto the DB
- Return type:
None
- data_in_memory: tp.List[KVTuple]¶
- property first_key_in_memory: Optional[K]¶
- Returns:
key of the first element stored in memory
- get_archive(start, stop)¶
Get some historic data that is kept both in the DB and in the memory
- Parameters:
start (K) – starting key (included)
stop (K) – stopping key (included)
- Returns:
a iterator of KVTuple
- Return type:
Iterator[Tuple[K, V]]
- get_latest_value()¶
Get the piece of data that was added here last
- Returns:
a tuple of (key, value)
- Raises:
ValueError – no data in series
- Return type:
Tuple[K, V]
- on_new_data(key, value)¶
Called by the user when there’s new data gathered.
Key must be greater than start entry
- Parameters:
key (K) – key of the new data
value (V) – value of the new data
- Raises:
ValueError – key was not larger than current stop entry
- Return type:
None
- on_sync_request(maximum_entries=inf)¶
Return an iterator that will provide the source of the data for synchronization.
This will preferentially start from the first value, so as to keep values synchronized in-order.
- Parameters:
maximum_entries (Optional[int]) –
- Returns:
an iterator of (KVTuple) that should be synchronized against the server
- Raises:
ValueError – nothing to synchronize!
- Return type:
Iterator[Tuple[K, V]]
- on_synced_up_to(key)¶
Called when data was successfully synced up to key included
- Parameters:
key (K) – maximum key synchronized
- Return type:
None
- span_to_keep_in_db: K¶
- span_to_keep_in_memory: K¶
- property start_entry: Optional[K]¶
- property stop_entry: Optional[K]¶
- sync_to_db()¶
Make sure that everything’s that in memory in also stored in the DB.
- Return type:
None
- property synced_up_to: Optional[K]¶
satella.coding.structures.tuples module¶
- class satella.coding.structures.tuples.Vector(iterable=(), /)¶
Bases:
tuple
A tuple that allows place-wise addition and subtraction of it’s entries, and also tuple-times-float operations
Ie.
>>> a = Vector((2, 3)) >>> b = Vector((1, 2)) >>> assert a-b == Vector((1, 1)) >>> assert a*2 == Vector((4, 6))
satella.coding.structures.typednamedtuple module¶
- satella.coding.structures.typednamedtuple.typednamedtuple(cls_name, *arg_name_type)¶
Returns a new subclass of tuple with named fields. Fields will be coerced to type passed in the pair, if they are not already of given type.
Parameters are tuples of (field name, class/constructor as callable/1)
For example:
>>> tnt = typednamedtuple('tnt', ('x', float), ('y', float)) >>> a = tnt('5.0', y=2)
a.x is float, a.y is float too
- Parameters:
cls_name (str) –
arg_name_type (type) –
- Return type:
Type[Tuple]
satella.coding.structures.zip_dict module¶
- class satella.coding.structures.zip_dict.SetZip(*args)¶
Bases:
object
An object which zips a bunch of sets together.
Ie. checks for inclusion by checking each set.
Also supports len and iteration protocol.
You can also add extra sets:
>>> c = SetZip() >>> c += set([1, 2, 3])
Provided arguments must implement contains, length and iter.
- Parameters:
args (set) –
Module contents¶
- class satella.coding.structures.CacheDict(stale_interval, expiration_interval, value_getter, value_getter_executor=None, cache_failures_interval=None, time_getter=<built-in function monotonic>, default_value_factory=None)¶
Bases:
Mapping
[K
,V
]A dictionary that you can use as a cache.
The idea is that you might want to cache some values, and they might be stale after some interval (after which they will need to be refreshed), and they will expire after some other interval (after which a call to get them will block until they are refreshed), but stale values are safe to serve from memory, while expired values are not and the dict will need to block until they are available.
If a stale value is read, a refresh is scheduled in the background for it. If an expired value is read, it will block until the result is available. Else, the value is served straight from fast memory.
Note that value_getter raising KeyError is not cached, so don’t use this cache for situations where misses are frequent.
- Parameters:
stale_interval (Union[float, int, str]) – time in seconds after which an entry will be stale, ie. it will be served from cache, but a task will be launched in background to refresh it. Note that this will accept time-like strings eg. 23m.
expiration_interval (Union[float, int, str]) – time in seconds after which an entry will be ejected from dict, and further calls to get it will block until the entry is available. Note that this will accept time-like strings eg. 23m.
value_getter (Callable[[K], V]) – a callable that accepts a key, and returns a value for given entry. If value_getter raises KeyError, then given entry will be evicted from the cache
value_getter_executor (Optional[Executor]) – an executor to execute the value_getter function in background. If None is passed, a ThreadPoolExecutor will be used with max_workers of 4.
cache_failures_interval (Optional[Union[float, int, str]]) – if any other than None is defined, this is the timeout for which failed lookups will be cached. By default they won’t be cached at all. Note that this will accept time-like strings eg. 23m.
time_getter (Callable[[], float]) – a routine used to get current time in seconds
default_value_factory (Optional[Callable[[], V]]) – if given, this is the callable that will return values that will be given to user instead of throwing KeyError. If not given (default), KeyError will be thrown
- feed(key, value, timestamp=None)¶
Feed this data into the cache
- Parameters:
key (K) –
value (V) –
timestamp (Optional[float]) –
- get_value_block(key)¶
Get a value using value_getter. Block until it’s available. Store it into the cache.
- Raises:
KeyError – the value is not present at all
- Parameters:
key (K) –
- Return type:
V
- has_info_about(key)¶
Is provided key cached, or failure about it is cached?
- Parameters:
key (K) –
- Return type:
bool
- invalidate(key)¶
Remove all information about given key from the cache
Syntactic sugar for:
>>> try: >>> del self[key] >>> except KeyError: >>> pass
- Parameters:
key (K) –
- Return type:
None
- schedule_a_fetch(key)¶
Schedule a value refresh for given key
- Parameters:
key (K) – key to schedule the refresh for
- Returns:
future that was queued to ask for given key
- Return type:
Future
- class satella.coding.structures.ComparableAndHashableBy¶
Bases:
object
A mix-in. Provides comparision (lt, gt, ge, le, eq) and hashing by a field of this class. Hashing is done by invoking hash() on the value of the field, and comparison is done by directly comparing given field’s values.
Example:
>>> class Vector(ComparableAndHashableBy): >>> _COMPARABLE_BY = 'length' >>> @property >>> def length(self): >>> return 0 >>> >>> assert Vector() > Vector()
- class satella.coding.structures.ComparableAndHashableByInt¶
Bases:
object
A mix-in. Provides comparison (lt, gt, ge, le, eq) and hashing by __int__ of this class.
- class satella.coding.structures.ComparableAndHashableByStr¶
Bases:
object
A mix-in. Provides comparision (lt, gt, ge, le, eq) and hashing by __str__ of this class. Also, will make this class equal to strings that return the same value.
- class satella.coding.structures.ComparableEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Bases:
Enum
An enum whose compare will try to convert value you compare it against to its instance. Handly for writing code like:
>>> a = 'test' >>> class Enum(ComparableEnum): >>> A = 'test' >>> assert Enum.A == a
Comparison order doesn’t matter, so either are True:
>>> Enum.A == 'test' >>> 'test' == Enum.A
Note, however, that following won’t work:
>>> 'test' in (Enum.A, )
You can even compare enums across classes
>>> class A(ComparableEnum): >>> A = 1 >>> class B(ComparableEnum): >>> A = 1 >>> assert A.A == B.A
- class satella.coding.structures.ComparableIntEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Bases:
HashableIntEnum
An enum.IntEnum that implements comparison, stemming from its values, as well as hashability.
It it has an int() method, it’s fair game as comparison argument for this class.
- class satella.coding.structures.CountingDict(set_to_count=())¶
Bases:
object
A dictionary to quickly count the amount of elements in a set
- Parameters:
set_to_count (Sequence) – a sequence, whose elements should be counted. The elements should support being keys in a dictionary.
- clear()¶
- count(elem, pieces=1)¶
Add an instance of elem to the set
- Parameters:
elem – instance to add
pieces (int) – amount to add
- Return type:
None
- dct¶
- items()¶
- keys()¶
- values()¶
- class satella.coding.structures.DBStorage¶
Bases:
object
An abstract implementation of the storage class provided to
SyncableDroppable
This serves as satella’s hook into your database infrastructure.
- abstract delete(key)¶
Called by SyncableDroppable when there’s a need to remove target key
- Parameters:
key (K) – key to remove
- Return type:
None
- abstract iterate(starting_key)¶
Return an iterator iterating from provided starting key to the end of the values, as read from the database.
This may block for a while.
This iterator will be closed upon no longer being necessary.
- Parameters:
starting_key (Optional[K]) – starting key, included, or None for iterate from the start
- Returns:
an iterator from provided key (included) to the end of the range
- Return type:
Iterator[Tuple[K, V]]
- abstract on_change_start_entry(start_entry)¶
Called by SyncableDroppable when starting entry (earliest entry encountered both in the DB and is memory) is changed.
- Parameters:
start_entry (Optional[K]) – new value of start entry or None if there are no entries at all
- Return type:
None
- abstract on_change_stop_entry(stop_entry)¶
Called by SyncableDroppable when stopping entry (earliest entry encountered both in the DB and is memory) is changed.
- Parameters:
stop_entry (Optional[K]) – new value of stop entry or None if there are no entries at all
- Return type:
None
- abstract on_change_synced_up_to(synced_up_to)¶
Called by SyncableDroppable when synced up to (earliest timestamp synced) is changed.
- Parameters:
synced_up_to (Optional[K]) – new value of synced up to
- Return type:
None
- abstract put(key, value)¶
Put given value to storage at given key.
This may block for a while.
- Parameters:
key (K) – key to store
value (V) – value to store
- Return type:
None
- class satella.coding.structures.DefaultDict¶
Bases:
defaultdict
A collections.defaultdict that does not store in itself empty values that it generates
- class satella.coding.structures.DictObject(*args, **kwargs)¶
Bases:
dict
A dictionary wrapper that can be accessed by attributes.
Note that it still remains a completely valid dictionary!
You can use keys different than strings, but they will be inaccessible as attributes, and you will have to do subscription to get them.
Eg:
>>> a = DictObject({'test': 5}) >>> self.assertEqual(a.test, 5) >>> self.assertEqual(a['test'], 5)
- is_valid_schema(schema=None, **kwarg_schema)¶
Check if this dictionary conforms to particular schema.
Schema is either a Descriptor, or a JSON-based schema. See satella.configuration.schema for details. Schema can be passed as well using kwargs. Note that the schema argument will be ignored if kwargs are passed.
- Parameters:
schema (Optional[Union[Descriptor, Dict]]) – schema to verify against
- Returns:
whether is conformant
- Return type:
bool
- class satella.coding.structures.DictionaryEQAble¶
Bases:
object
A class mix-in that defines __eq__ and __ne__ to be:
both the same exact type (so subclassing won’t work)
have the exact same __dict__
- class satella.coding.structures.DictionaryView(master_dict, *rest_of_dicts, propagate_deletes=True, assign_to_same_dict=True)¶
Bases:
MutableMapping
[K
,V
]A view on a multiple dictionaries. If key isn’t found in the first dictionary, it is looked up in another. Use like:
>>> dv = DictionaryView({1:2, 3:4}, {4: 5, 6: 7}) >>> assert dv[4] == 5 >>> del dv[1] >>> assertRaises(KeyError, lambda: dv.__delitem__(1))
Deprecated since version 2.14.22: Use ChainMap instead.
- Parameters:
master_dict (tp.Dict[K, V]) – First dictionary to look up. Entries made via __setitem__ will be put here.
rest_of_dicts (tp.Dict[K, V]) – Remaining dictionaries
propagate_deletes (bool) – Whether to delete given key from the first dictionary that it is found. Otherwise it will be only deleted from the master_dict. Also, if this is set to False, on deletion, if the key isn’t found in master dictionary, deletion will KeyError.
assign_to_same_dict (bool) – whether updates done by __setitem__ should be written to the dictionary that contains that key. If not, all updates will be stored in master_dict. If this is True, updates made to keys that are not in this dictionary will go to master_dict.
- assign_to_same_dict¶
- dictionaries¶
- master_dict¶
- propagate_deletes¶
- class satella.coding.structures.DirtyDict(*args, **kwargs)¶
Bases:
MutableMapping
[K
,V
]A dictionary that has also a flag called .dirty that sets to True if the dictionary has been changed since that flag was last cleared.
Setting the dict with the value that it already has doesn’t count as dirtying it. Note that such changes will not be registered in the dict!
All arguments and kwargs will be passed to dict’s constructor.
- clear_dirty()¶
Clears the dirty flag
- Return type:
None
- copy_and_clear_dirty()¶
Returns a copy of this data and sets dirty to False
- Returns:
a plain, normal Python dictionary is returned
- Return type:
Dict[K, V]
- swap_and_clear_dirty()¶
Returns this data, clears self and sets dirty to False
After this is called, this dict will be considered empty.
- Returns:
a plain, normal Python dictionary is returned
- Return type:
Dict[K, V]
- class satella.coding.structures.ExclusiveWritebackCache(write_method, read_method, delete_method=None, executor=None, no_concurrent_executors=None, store_key_errors=True)¶
Bases:
Generic
[K
,V
]A dictionary implementing an exclusive write-back cache. By exclusive it is understood that only this object will be modifying the storage.
- Parameters:
write_method (Callable[[K, V], None]) – a blocking callable (key, value) that writes the value to underlying storage.
read_method (Callable[[K], V]) – a blocking callable (key) -> value that retrieves the piece of data from the cache, or throws a KeyError to signal that the value does not exist
delete_method (Optional[Callable[[K], None]]) – optional, a blocking callable (key) that erases the data from the storage. If not given, it will be a TypeError to delete the data from this storage
executor (Optional[Executor]) – an executor to execute the calls with. If None (default) is given, a ThreadPoolExecutor with 4 workers will be created
no_concurrent_executors (Optional[int]) – number of concurrent jobs that the executor is able to handle. This is used by sync()
store_key_errors (bool) – whether to remember KeyErrors raised by read_method
- cache¶
- cache_lock¶
- delete_method¶
- executor¶
- get_queue_length()¶
Return current amount of entries waiting for writeback
- Return type:
int
- in_cache¶
- no_concurrent_executors¶
- operations¶
- read_method¶
- store_key_errors¶
- sync(timeout=None)¶
Wait until current tasks are complete.
- Parameters:
timeout (Optional[float]) – timeout to wait. None means wait indefinitely.
- Raises:
WouldWaitMore – if timeout has expired
- Return type:
None
- write_method¶
- class satella.coding.structures.ExpiringEntryDict(expiration_timeout, *args, time_getter=<built-in function monotonic>, external_cleanup=False, **kwargs)¶
Bases:
Monitor
,MutableMapping
[K
,V
],Cleanupable
A dictionary whose entries expire automatically after a predefined period of time.
Note that cleanup is invoked only when iterating over the dicts, or automatically if you specify external_cleanup to be True each approximately 5 seconds.
Note that it’s preferential to
satella.coding.concurrent.Monitor.acquire()
it if you’re using an external cleanup thread, because the dict may mutate at any time.- Parameters:
expiration_timeout (float) – number of seconds after which entries will expire
time_getter (Callable[[], float]) – a callable/0 that returns the current timestamp
external_cleanup (bool) – whether to spawn a single thread that will clean up the dictionary. The thread is spawned once per program, and no additional threads are spawned for next dictionaries.
All args and kwargs will be passed to a dict, which will be promptly added to this dictionary.
You need to invoke this at your constructor You can also use it to release locks of other objects.
- cleanup()¶
Remove entries that are later than given time
- Return type:
None
- get_timestamp(key)¶
Return the timestamp at which given key was inserted in the dict
- Raises:
KeyError – key not found in the dictionary
- Parameters:
key (K) –
- Return type:
float
- class satella.coding.structures.HashableIntEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Bases:
IntEnum
An enum.IntEnum that implements hashability, stemming from its values, as well as hashability
- class satella.coding.structures.HashableMixin¶
Bases:
object
Make a class hashable by its ID.
Just remember to add the following to your class definition if you’re overriding __eq__:
>>> class MyClass(HashableMixin): >>> __hash__ = HashableMixin.__hash__
- class satella.coding.structures.HashableWrapper(object_to_wrap, wrap_operations=False)¶
Bases:
Proxy
A class that makes given objects hashable by their id, if the object is already not hashable.
Also overrides __eq__
Note that this class will return a proxy to the object, and not the object itself.
Use like:
>>> a = {1:2, 3:4} >>> a = HashableWrapper(a) >>> hash(a)
- Parameters:
object_to_wrap (T) –
wrap_operations (bool) –
- class satella.coding.structures.Heap(from_list=None)¶
Bases:
UserList
,Generic
[T
]Sane heap as object - not like heapq.
Goes from lowest-to-highest (first popped is smallest). Standard Python comparision rules apply.
Not thread-safe
- Parameters:
from_list (Optional[Iterable[T]]) –
- filter_map(filter_fun=None, map_fun=None)¶
- Get only items that return True when condition(item) is True. Apply a
transform: item’ = item(condition) on
the rest. Maintain heap invariant.
- Parameters:
filter_fun (Optional[Callable[[T], bool]]) –
map_fun (Optional[Callable[[T], Any]]) –
- iter_ascending()¶
Return an iterator returning all elements in this heap sorted ascending. State of the heap is not changed
- Return type:
Iterable[T]
- iter_descending()¶
Return an iterator returning all elements in this heap sorted descending. State of the heap is not changed.
This loads all elements of the heap into memory at once, so be careful.
- Return type:
Iterable[T]
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
T
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
T
- push(item, *args)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (T) –
- Return type:
None
- push_many(items)¶
Put multiple items on the heap. Faster than
>>> for item in items: >>> heap.push(item)
- Param:
an iterable with items to put
- Parameters:
items (Iterable[T]) –
- Return type:
None
- class satella.coding.structures.Immutable(*args, **kwargs)¶
Bases:
object
A mix-in to make your classes immutable.
You can assign normally using your constructor.
>>> class Test(Immutable): >>> def __init__(self): >>> self.attribute = 'value'
- class satella.coding.structures.KeyAwareDefaultDict(factory_function, *args, **kwargs)¶
Bases:
MutableMapping
[K
,V
]A defaultdict whose factory function accepts the key to provide a default value for the key
- Parameters:
factory_function (tp.Callable[[K], V]) – a callable that accepts a single argument, a key, for which it is to provide a default value
- class satella.coding.structures.LRU¶
Bases:
Generic
[T
]A class to track least recently used objects.
- add(item)¶
- Parameters:
item (T) –
- Return type:
None
- get_item_to_evict()¶
Return a least recently used object
- Return type:
T
- mark_as_used(item)¶
- Parameters:
item (T) –
- remove(item)¶
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.structures.LRUCacheDict(*args, max_size=100, **kwargs)¶
Bases:
CacheDict
[K
,V
]A dictionary that you can use as a cache with a maximum size, items evicted by LRU policy.
- Parameters:
max_size (int) – maximum size
- evict()¶
- feed(key, value, timestamp=None)¶
Feed this data into the cache
- Parameters:
key (K) –
value (V) –
timestamp (Optional[float]) –
- get_value_block(key)¶
Get a value using value_getter. Block until it’s available. Store it into the cache.
- Raises:
KeyError – the value is not present at all
- Parameters:
key (K) –
- Return type:
V
- invalidate(key)¶
Remove all information about given key from the cache
Syntactic sugar for:
>>> try: >>> del self[key] >>> except KeyError: >>> pass
- Parameters:
key (K) –
- Return type:
None
- make_room()¶
Assure that there’s place for at least one element
- Return type:
None
- class satella.coding.structures.NotEqualToAnything¶
Bases:
object
A class that defines __eq__ and __ne__ to be always False and True respectively
When exploiting this class please make sure that it gets to be the first operand in the comparison, so that it’s __eq__ and __ne__ gets called!
- class satella.coding.structures.OmniHashableMixin¶
Bases:
object
A mix-in. Provides hashing and equal comparison for your own class using specified fields.
Example of use:
>>> class Point2D(OmniHashableMixin): >>> _HASH_FIELDS_TO_USE = ('x', 'y') >>> def __init__(self, x, y): >>> ...
and now class Point2D has defined __hash__ and __eq__ by these fields. Do everything in your power to make specified fields immutable, as mutating them will result in a different hash.
This will also check if the other value is an instance of this instance’s class.
_HASH_FIELDS_TO_USE can be also a single string, in this case a single field called by this name will be taken.
Note that if you’re explicitly providing __eq__ in your child class, you will be required to insert:
>>> __hash__ = OmniHashableMixin.__hash__
for this to work in your class
- class satella.coding.structures.OnStrOnlyName¶
Bases:
object
A mix-in to add the following functionality to your class.
tl;dr - the name will be used instead of ClassName.name.
>>> from enum import Enum >>> class MyEnum(OnStrOnlyName, Enum): >>> A = 0 >>> B = 1 >>> C = 'test' >>> assert str(MyEnum.A) == 'A' >>> assert str(MyEnum.B) == 'B' >>> assert str(MyEnum.C) == 'test'
- class satella.coding.structures.Proxy(object_to_wrap, wrap_operations=False)¶
Bases:
Generic
[T
]A base class for classes that try to emulate some other object.
They will intercept all calls and place them on the target object.
Note that in-place operations will return the Proxy itself, whereas simple addition will shed this proxy, returning object wrapped plus something.
Only __isinstance__ check is not overridden, due to Python limitations.
Please access this object in your descendant classes via
>>> getattr(self, '_Proxy__obj')
Please note that this class does not overload the descriptor protocol, not the pickle interface!
Note that this overloads __repr__, __str__ and __dir__, which may prove confusing. Handle this in your descendant classes.
If wrap_operations is set, the following code will be executed on the result:
>>> a = a.__add__(b) >>> return self.__class__(a)
Wrapped operations include all arithmetic operations and bitwise operations, except for divmod, but including concat, rounding, truncing and ceiling.
- Parameters:
object_to_wrap (T) – object to wrap
wrap_operations (bool) – whether results of operations returning something else should be also proxied.
- class satella.coding.structures.PushIterable(iterable)¶
Bases:
Generic
[T
]An iterable that you can add elements to it’s head, and they will be popped before this iterable is consumed when a pop is called.
Example:
a = PushIterable([1, 2, 3]) assert a.pop() == 1 a.push(0) assert a.pop() == 0 assert a.pop() == 2 a.push(0) a.push(1) assert a.pop() == 1 assert a.pop() == 0 a.push_left(0) a.push_left(1) assert a.pop() == 0 assert a.pop() == 1 assert a.pop() == 3 assertRaises(StopIteration, a.pop)
- Parameters:
iterable (Iterable[T]) –
- pop()¶
Return next element from the stack :return: a next element
- Return type:
T
- push(item)¶
Push an item so that next pop will retrieve this item
- Parameters:
item (T) – item to push
- Return type:
None
- push_left(item)¶
Push an item so that last pop from internal buffer will retrieve this item
- Parameters:
item – item to push
- class satella.coding.structures.Ranking(items, key)¶
Bases:
Generic
[T
]A set of objects with them being ranked by their single property with the assumption that this property changes only when we want to.
Positions are counted from 0, where 0 has the least key value.
Essentially, this is a SortedList with the option to query at which position can be given element found.
Example usage:
>>> Entry = collections.namedtuple('Entry', ('key', )) >>> e1 = Entry(2) >>> e2 = Entry(3) >>> e3 = Entry(5) >>> ranking = Ranking([e1, e2, e3], lambda e: e.key) >>> assert ranking[0] == e1 # Get the first element >>> assert ranking[-1] == e3 # Get the last element >>> assert ranking.get_position_of(e1) == 0
- Parameters:
items (List[T]) –
key (Callable[[T], int]) –
- add(item)¶
Add a single element to the ranking and _recalculate it
- Parameters:
item (T) –
- Return type:
None
- calculate_ranking_for(item)¶
- Parameters:
item (T) –
- Return type:
int
- element_to_position: Dict[int, int]¶
- get_position_of(item)¶
Return the position in the ranking of element item
- Parameters:
item (T) – element to return the position for
- Returns:
position
- Raises:
ValueError – this element is not in the ranking
- Return type:
int
- get_sorted()¶
Return all the elements sorted from the least key value to the highest key value
- Return type:
Iterator[T]
- items¶
- key¶
- ranking: SortedList[T]¶
- remove(item)¶
Remove a single element from the ranking and _recalculate it
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.structures.ReprableMixin¶
Bases:
object
A sane __repr__ default.
This takes the values for the __repr__ from repr’ing list of fields defined as class property _REPR_FIELDS.
Set an optional class property of _REPR_FULL_CLASSNAME for __repr__ to output the repr alongside the module name.
Example:
>>> class Test(ReprableMixin): >>> _REPR_FIELDS = ('v', ) >>> def __init__(self, v, **kwargs): >>> self.v = v >>> >>> assert repr(Test(2)) == "Test(2)" >>> assert repr(Test('2') == "Test('2')")
- class satella.coding.structures.SelfCleaningDefaultDict(default_factory, background_maintenance=True, *args, **kwargs)¶
Bases:
Monitor
,MutableMapping
[K
,V
],Cleanupable
A defaultdict with the property that if it detects that a value is equal to it’s default value, it will automatically remove it from the dict.
Please note that this will spawn a cleanup thread in the background, one per program. The thread is shared between
satella.coding.structures.SelfCleaningDefaultDict
andsatella.coding.structures.ExpiringEntryDict
It is preferable to
satella.coding.concurrent.Monitor.acquire()
it before iterating, since it is cleaned up both by __iter__ and by an external worker thread, so it’s important to acquire it, because it can mutate in an undefined way.Note that if you access a key which does not exist, and background_maintenance is False, a default value will be created and inserted into the dictionary. This is the only time that the dictionary will hold values that are equal to default.
- Parameters:
default_factory (Callable[[], V]) – a callable/0 that will return an object if it doesn’t already exist
background_maintenance (bool) – whether to spawn a background thread. This is required if dictionary values can change their value between inserts.
All args and kwargs will be passed to a dict, which will be promptly added to this dictionary.
You need to invoke this at your constructor You can also use it to release locks of other objects.
- cleanup()¶
- Return type:
None
- class satella.coding.structures.SetHeap(from_list=None)¶
Bases:
Heap
A heap with additional invariant that no two elements are the same.
Note that elements you insert in this must be eq-able and hashable, ie. you can put them in a dict.
Optimized for fast insertions and fast __contains__
#notthreadsafe
- Parameters:
from_list (Optional[Iterable[T]]) –
- filter_map(filter_fun=None, map_fun=None)¶
- Get only items that return True when condition(item) is True. Apply a
transform: item’ = item(condition) on
the rest. Maintain heap invariant.
- Parameters:
filter_fun (Optional[Callable[[T], bool]]) –
map_fun (Optional[Callable[[T], Any]]) –
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
T
- push(item)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (T) –
- push_many(items)¶
Not that much faster than inserting anything by hand
- Parameters:
items (Iterable[T]) –
- Return type:
None
- class satella.coding.structures.SetZip(*args)¶
Bases:
object
An object which zips a bunch of sets together.
Ie. checks for inclusion by checking each set.
Also supports len and iteration protocol.
You can also add extra sets:
>>> c = SetZip() >>> c += set([1, 2, 3])
Provided arguments must implement contains, length and iter.
- Parameters:
args (set) –
- satella.coding.structures.Singleton(cls)¶
Make a singleton out of decorated class.
Usage:
>>> @Singleton >>> class MyClass: >>> ...
- satella.coding.structures.SingletonWithRegardsTo(num_args, weak_refs=False)¶
Make a memoized singletion depending on the arguments.
A dictionary is made (first N arguments => class instance) and such is returned. Please take care to ensure that a tuple made out of first num_args can be used as a dictionary key (ie. is both hashable and __eq__-able).
Usage:
>>> @SingletonWithRegardsTo(num_args=1) >>> class MyClass: >>> def __init__(self, device_id: str): >>> ... >>> a = MyClass('dev1') >>> b = MyClass('dev2') >>> c = MyClass('dev1') >>> assert a is c >>> assert b is not c
- Parameters:
num_args (int) – number of arguments to consume
weak_refs (bool) – if True, then singleton will be stored within a weak dictionary, so that it cleans up after itself when the values are gone.
Warning
If you set weak_refs to False and have a potentially unbounded number of argument values, you better watch out for the memory usage.
- class satella.coding.structures.SliceableDeque(*args, **kwargs)¶
Bases:
MutableSequence
,Generic
[T
]A collections.deque that supports slicing.
Just note that it will return a p_gen upon being sliced!
- deque¶
- insert(i, item)¶
S.insert(index, value) – insert value before index
- Parameters:
i (int) –
item (T) –
- class satella.coding.structures.SortedList(items=(), key=<function SortedList.<lambda>>)¶
Bases:
Generic
[T
]An always-sorted sort of a set
It is assumed that keys of constituent elements don’t change.
list[0] will have the smallest element, and list[-1] the biggest.
Addition is O(n) worst-case, so is deletion.
- Parameters:
items (SliceableDeque[T]) – items to construct the list with
key (Callable[[T], int]) – a callable[T]->int that builds the key of the sort
- add(other)¶
Add an element. Returns the index at which it was inserted.
- Parameters:
other (T) – element to insert
- Returns:
index that the entry is available now at
- Return type:
int
- extend(elements)¶
Adds multiple elements to this list
- Parameters:
elements (Iterable[T]) –
- index(other)¶
Return index at which given value has been placed
- Parameters:
other (T) –
- Return type:
int
- items: SliceableDeque[T]¶
- key: Callable[[T], int]¶
- keys: deque[int]¶
- pop()¶
Return the highest element, removing it from the list
- Return type:
T
- popleft()¶
Return the smallest element, removing it from the list
- Return type:
T
- remove(other)¶
Remove an element from the list
- Parameters:
other (T) – element to remove
- Raises:
ValueError – element not in list
- Return type:
None
- class satella.coding.structures.SparseMatrix(matrix_data=None)¶
Bases:
Generic
[T
]A matrix of infinite size, that supports assignments.
Set elements like this:
>>> sm[1, 2] = 5 >>> sm[:,1] = [5] >>> sm[1,:] = [5] >>> sm[:,:] = [[5]]
where the first argument is the number of the column, counted from 0, and the second one is the number of the row, also counted from 0
Note that custom slicing (ie. slices which are not :) will not be supported. Negative indices are supported.
Undefined elements are considered to be of value None.
Iterating over this matrix will yield it’s consecutive rows.
You can use the constructor in following way:
>>> sm = SparseMatrix([[1, 2], [3, 4]])
to construct a matrix that looks like
|1 2| |3 4|
- Parameters:
matrix_data (Optional[List[List[T]]]) –
- append_row(y)¶
Append a row to the bottom of the matrix
- Parameters:
y (Iterable[T]) – iterable with consequent columns
- Return type:
None
- clear()¶
Clear the contents of the sparse matrix
- Return type:
None
- property columns: int¶
Return the amount of columns
- delete_row(row_no)¶
Delete a row with specified number
- Parameters:
row_no (int) – number of the row to delete
- Return type:
None
- classmethod from_iterable(y)¶
Construct a sparse matrix given a row-first iterable. That iterable must return another iterable, that will yield values for given column.
- Parameters:
y (Iterable[Iterable[T]]) – an iterable describing the sparse matrix
- Returns:
a sparse matrix object
- get_neighbour_coordinates(col, row, include_diagonals=True)¶
Return an iterator of coordinates to points neighbouring given point. :param col: column :param row: row :param include_diagonals: whether to include points having only a single point in common :return: an iterable of coordinates of neighbouring points. An iterator of tuple
(col, row)
- Parameters:
col (int) –
row (int) –
include_diagonals (bool) –
- Return type:
Iterator[Tuple[int, int]]
- get_row(row_no)¶
Return a single row of provided number.
The returned array has the same length as .columns
- Parameters:
row_no (int) – row number, numbered from 0
- Return type:
List[T]
- known_column_count¶
- max()¶
Return maximum element.
None elements will be ignored.
- Return type:
T
- min()¶
Return minimum element.
None elements will be ignored.
- Return type:
T
- no_cols¶
- no_rows¶
- property rows: int¶
Return the amount of rows
- rows_dict¶
- shoot()¶
Insert an empty cell between current cells. So the matrix which looked like [[1, 2], [3, 4]] will now look like [[1, None, 2], [None, None, None], [3, None, 4]]
- Return type:
- class satella.coding.structures.StrEqHashableMixin¶
Bases:
object
A mix-in that outfits your class with an __eq__ and __hash__ operator that take their values from __str__ representation of your class.
- class satella.coding.structures.Subqueue¶
Bases:
Generic
[T
]Or a named queue is a collection of thread safe queues identified by name
- assert_queue(queue_name)¶
Assure that we have a queue with a particular name in the dictionary
- Parameters:
queue_name (str) –
- Return type:
None
- get(queue_name, block=True, timeout=None)¶
Same semantics at queue.Queue.get
- Parameters:
queue_name (str) –
- Return type:
T
- get_any()¶
Return any message with name of the queue.
This assumes block=False
- Returns:
a tuple of (queue_name, object)
- Raises:
queue.Empty()
- Return type:
Tuple[str, T]
- put(queue_name, obj)¶
Same semantics as queue.Queue.put
- Parameters:
queue_name (str) –
obj (T) –
- Return type:
None
- qsize()¶
Calculate the total of entries
- Return type:
int
- class satella.coding.structures.SyncableDroppable(db_storage, start_entry, stop_entry, synced_up_to, span_to_keep_in_memory, span_to_keep_in_db)¶
Bases:
RMonitor
,Generic
[K
,V
]A thread-safe class representing some single time series, which needs to be synchronized with some server, and may be too large to keep in memory. Moreover, after the sync we need to retain a part of the time series in memory for future requests. Only series past some timestamp may be deleted.
For brevity, this will refer to keys as timestamps. The keys must be __eq__able, comparable and subtractable.
A rule is that an item can never be both in memory and in the DB.
- Parameters:
db_storage (DBStorage) – a DBStorage implementation of your own provision, that serves as class’ interface with the database
start_entry (Optional[K]) – earliest timestamp stored or None if no data is stored
stop_entry (Optional[K]) – latest timestamp stored or None if no data is stored
synced_up_to (Optional[K]) – timestamp of the last entry synchronized or None if no data is stored
span_to_keep_in_memory (int) – key span to keep in memory. Entries earlier than difference of the latest key and this will be dropped from memory, onto the DB. Can’t be false.
span_to_keep_in_db (int) – key span to keep on disk. Entries earlier than difference of the latest key and this will be dropped from the DB. Can’t be false.
Note
Note that proper handling of maximum spans requires periodical calls to
cleanup()
You need to invoke this at your constructor You can also use it to release locks of other objects.
- cleanup()¶
Make sure that everything’s that in memory and the DB conforms to span_to_keep_in_db and span_to_keep_in_memory.
This may block for a while.
Warning
This bugs out for now. For now, a UserWarning will be raised and nothing will be done.
- Return type:
None
- cleanup_keep_in_db()¶
Clear up the database to conform to our span_to_keep_in_db
- Return type:
None
- cleanup_keep_in_memory()¶
Eject values from memory that should reside in the DB onto the DB
- Return type:
None
- data_in_memory: tp.List[KVTuple]¶
- property first_key_in_memory: Optional[K]¶
- Returns:
key of the first element stored in memory
- get_archive(start, stop)¶
Get some historic data that is kept both in the DB and in the memory
- Parameters:
start (K) – starting key (included)
stop (K) – stopping key (included)
- Returns:
a iterator of KVTuple
- Return type:
Iterator[Tuple[K, V]]
- get_latest_value()¶
Get the piece of data that was added here last
- Returns:
a tuple of (key, value)
- Raises:
ValueError – no data in series
- Return type:
Tuple[K, V]
- on_new_data(key, value)¶
Called by the user when there’s new data gathered.
Key must be greater than start entry
- Parameters:
key (K) – key of the new data
value (V) – value of the new data
- Raises:
ValueError – key was not larger than current stop entry
- Return type:
None
- on_sync_request(maximum_entries=inf)¶
Return an iterator that will provide the source of the data for synchronization.
This will preferentially start from the first value, so as to keep values synchronized in-order.
- Parameters:
maximum_entries (Optional[int]) –
- Returns:
an iterator of (KVTuple) that should be synchronized against the server
- Raises:
ValueError – nothing to synchronize!
- Return type:
Iterator[Tuple[K, V]]
- on_synced_up_to(key)¶
Called when data was successfully synced up to key included
- Parameters:
key (K) – maximum key synchronized
- Return type:
None
- span_to_keep_in_db: K¶
- span_to_keep_in_memory: K¶
- property start_entry: Optional[K]¶
- property stop_entry: Optional[K]¶
- sync_to_db()¶
Make sure that everything’s that in memory in also stored in the DB.
- Return type:
None
- property synced_up_to: Optional[K]¶
- class satella.coding.structures.TimeBasedHeap(default_clock_source=<built-in function monotonic>)¶
Bases:
Heap
A heap of items sorted by timestamps.
It is easy to ask for items, whose timestamps are LOWER than a value, and easy to remove them.
Can be used to implement a scheduling service, ie. store jobs, and each interval query which of them should be executed. This loses time resolution, but is fast.
Can use current time with put/pop_less_than. Use default_clock_source to pass a callable:
time.time
time.monotonic
Default is time.monotonic
#notthreadsafe
Initialize an empty heap
- Parameters:
default_clock_source (Callable[[], Union[int, float]]) –
- get_timestamp(item)¶
Return the timestamp for given item
- Parameters:
item (T) –
- Return type:
Union[int, float]
- items()¶
Return an iterable, but WITHOUT timestamps (only items), in unspecified order
- Return type:
Iterable[Tuple[Union[int, float], T]]
- peek_closest()¶
Return the closest object to the execution deadline, but not discard it from the heap.
- Raises:
IndexError – the heap is empty
- Return type:
Tuple[Union[int, float], T]
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant.
The item will be a second part of the tuple
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
Tuple[Union[int, float], T]
- pop_less_than(less=None)¶
Return all elements less (sharp inequality) than particular value.
This changes state of the heap
- Parameters:
less (Optional[Union[int, float]]) – value to compare against. If left at default, it will be the default clock source specified at construction.
- Returns:
an Iterator
- Return type:
Iterator[Tuple[Union[int, float], T]]
- pop_timestamp(timestamp)¶
Get first item with given timestamp, while maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
timestamp (Union[int, float]) –
- Return type:
T
- put(timestamp_or_value, value=None)¶
Put an item on heap.
Pass timestamp, item or just an item for default time
- Parameters:
timestamp_or_value (Union[T, int, float]) –
value (Optional[T]) –
- Return type:
None
- remove(item)¶
Remove all things equal to item
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.structures.TimeBasedSetHeap(default_clock_source=None)¶
Bases:
Heap
A heap of items sorted by timestamps, with such invariant that every item can appear at most once.
Note that elements you insert in this must be eq-able and hashable, ie. you can put them in a dict.
It is easy to ask for items, whose timestamps are LOWER than a value, and easy to remove them.
Can be used to implement a scheduling service, ie. store jobs, and each interval query which of them should be executed. This loses time resolution, but is fast.
Can use current time with put/pop_less_than. Use default_clock_source to pass a callable:
time.time
time.monotonic
Default is time.monotonic
#notthreadsafe
Initialize an empty heap
- Parameters:
default_clock_source (Callable[[], Union[int, float]]) –
- get_timestamp(item)¶
Return the timestamp for given item
- Raises:
ValueError – item not found
- Parameters:
item (T) –
- Return type:
Union[int, float]
- items()¶
Return an iterator, but WITHOUT timestamps (only items), in unspecified order
- Return type:
Iterator[T]
- pop()¶
Return smallest element of the heap.
- Raises:
IndexError – on empty heap
- Return type:
Tuple[Union[int, float], T]
- pop_item(item)¶
Pop an item off the heap, maintaining the heap invariant.
The item will be a second part of the tuple
- Raises:
ValueError – element not found
- Parameters:
item (T) –
- Return type:
Tuple[Union[int, float], T]
- pop_less_than(less=None)¶
Return all elements less (sharp inequality) than particular value.
This changes state of the heap
- Parameters:
less (Optional[Union[int, float]]) – value to compare against
- Returns:
a Generator
- Return type:
Iterator[Tuple[Union[int, float], T]]
- pop_timestamp(timestamp)¶
Pop an arbitary object (in case there’s two) item with given timestamp, while maintaining the heap invariant
- Raises:
ValueError – element not found
- Parameters:
timestamp (Union[int, float]) –
- Return type:
T
- push(item)¶
Use it like:
>>> heap.push(3)
or:
>>> heap.push(4, myobject)
However this syntax is
Deprecated since version 2.25.5: It’s not legible
- Parameters:
item (Tuple[Union[int, float], T]) –
- Return type:
None
- put(timestamp_or_value, value=None)¶
Put an item on heap.
Pass timestamp, item or just an item for default time
- Parameters:
timestamp_or_value (Union[T, int, float]) –
value (Optional[T]) –
- Return type:
None
- remove(item)¶
Remove all things equal to item
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.structures.TwoWayDictionary(data=None, _is_reverse=False)¶
Bases:
MutableMapping
[K
,V
]A dictionary that keeps also a reverse_data mapping, allowing to look up keys by values.
Not thread-safe.
Example usage:
>>> twd = TwoWayDictionary() >>> twd[2] = 3 >>> self.assertEqual(twd.reverse[3], 2)
When you’re done using a given TwoWayDictionary, please call .done(). This will make it easier for the GC to collect the dictionaries.
You can also use the context manager to make the TwoWayDictionary clean up itself, eg.
>>> with TwoWayDictionary() as twd: >>> ... >>> # at this point twd is .done()
- Parameters:
data – data to generate the dict from
_is_reverse (bool) –
- Raises:
ValueError – on being given data from which it is impossible to construct a reverse mapping (ie. same value appears at least twice)
- data¶
- done()¶
Called when the user is done using given TwoWayDictionary.
Internally this will break the reference cycle, and enable Python GC to collect the objects.
- Return type:
None
- property reverse: MutableMapping[V, K]¶
Return a reverse mapping. Reverse mapping is updated as soon as an operation is done.
- reverse_data¶
- class satella.coding.structures.Vector(iterable=(), /)¶
Bases:
tuple
A tuple that allows place-wise addition and subtraction of it’s entries, and also tuple-times-float operations
Ie.
>>> a = Vector((2, 3)) >>> b = Vector((1, 2)) >>> assert a-b == Vector((1, 1)) >>> assert a*2 == Vector((4, 6))
- satella.coding.structures.apply_dict_object(v)¶
Apply DictObject() to every dict inside v.
This assumes that the only things that will be touched will be nested dicts and lists.
If you pass a non-dict and a non-list, they will be returned as is.
- Parameters:
v (Union[Any, Dict]) –
- Return type:
Union[DictObject, Any]
- satella.coding.structures.delete_singleton_for(x, *args)¶
Delete singleton for given arguments in a class decorated with SingletonWithRegardsTo
- Parameters:
x – class decorated with SingletonWithRegardsTo
args – arguments used in the constructor
- Return type:
None
- class satella.coding.structures.frozendict¶
Bases:
dict
A hashable dict with express forbid to change it’s values Both keys and values must be hashable in order for this dict to be hashable.
- update([E, ]**F) None. Update D from dict/iterable E and F. ¶
If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]
- satella.coding.structures.get_instances_for_singleton(x)¶
Obtain a list of arguments for which singletons exists for given class decorated with SingletonWithRegardsTo
- Parameters:
x – a class decorated with SingletonWithRegardsTo
- Returns:
a list of arguments
- Return type:
List[Tuple]
- satella.coding.structures.typednamedtuple(cls_name, *arg_name_type)¶
Returns a new subclass of tuple with named fields. Fields will be coerced to type passed in the pair, if they are not already of given type.
Parameters are tuples of (field name, class/constructor as callable/1)
For example:
>>> tnt = typednamedtuple('tnt', ('x', float), ('y', float)) >>> a = tnt('5.0', y=2)
a.x is float, a.y is float too
- Parameters:
cls_name (str) –
arg_name_type (type) –
- Return type:
Type[Tuple]