Structures¶

class satella.coding.structures.NotEqualToAnything¶

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!

You can also use the following singleton.

class satella.coding.structures.Vector(iterable=(), /)¶

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))

SetZip¶

class satella.coding.structures.SetZip(*args: set)¶

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.

PushIterable¶

class satella.coding.structures.PushIterable(iterable: Iterable[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)

pop() → T¶: Return next element from the stack :return: a next element

push(item: T) → None¶

Push an item so that next pop will retrieve this item

Parameters:: item – item to push

push_left(item)¶

Push an item so that last pop from internal buffer will retrieve this item

Parameters:: item – item to push

SyncableDroppable¶

class satella.coding.structures.SyncableDroppable(db_storage: DBStorage, start_entry: K | None, stop_entry: K | None, synced_up_to: K | None, span_to_keep_in_memory: int, span_to_keep_in_db: int)¶

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 – a DBStorage implementation of your own provision, that serves as class’ interface with the database
start_entry – earliest timestamp stored or None if no data is stored
stop_entry – latest timestamp stored or None if no data is stored
synced_up_to – timestamp of the last entry synchronized or None if no data is stored
span_to_keep_in_memory – 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 – 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()

cleanup() → None¶

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.

cleanup_keep_in_db() → None¶: Clear up the database to conform to our span_to_keep_in_db

cleanup_keep_in_memory() → None¶: Eject values from memory that should reside in the DB onto the DB

property first_key_in_memory: K | None¶

Returns:: key of the first element stored in memory

get_archive(start: K, stop: K) → Iterator[Tuple[K, V]]¶

Get some historic data that is kept both in the DB and in the memory

Parameters:

start – starting key (included)
stop – stopping key (included)

Returns:

a iterator of KVTuple

get_latest_value() → Tuple[K, V]¶

Get the piece of data that was added here last

Returns:: a tuple of (key, value)
Raises:: ValueError – no data in series

on_new_data(key: K, value: V) → None¶

Called by the user when there’s new data gathered.

Key must be greater than start entry

Parameters:

key – key of the new data
value – value of the new data

Raises:

ValueError – key was not larger than current stop entry

on_sync_request(maximum_entries: int | None = inf) → Iterator[Tuple[K, V]]¶

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 –
Returns:: an iterator of (KVTuple) that should be synchronized against the server
Raises:: ValueError – nothing to synchronize!

on_synced_up_to(key: K) → None¶

Called when data was successfully synced up to key included

Parameters:: key – maximum key synchronized

sync_to_db() → None¶: Make sure that everything’s that in memory in also stored in the DB.

class satella.coding.structures.DBStorage¶

An abstract implementation of the storage class provided to SyncableDroppable

This serves as satella’s hook into your database infrastructure.

abstract delete(key: K) → None¶

Called by SyncableDroppable when there’s a need to remove target key

Parameters:: key – key to remove

abstract iterate(starting_key: K | None) → Iterator[Tuple[K, V]]¶

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 – starting key, included, or None for iterate from the start
Returns:: an iterator from provided key (included) to the end of the range

abstract on_change_start_entry(start_entry: K | None) → None¶

Called by SyncableDroppable when starting entry (earliest entry encountered both in the DB and is memory) is changed.

Parameters:: start_entry – new value of start entry or None if there are no entries at all

abstract on_change_stop_entry(stop_entry: K | None) → None¶

Called by SyncableDroppable when stopping entry (earliest entry encountered both in the DB and is memory) is changed.

Parameters:: stop_entry – new value of stop entry or None if there are no entries at all

abstract on_change_synced_up_to(synced_up_to: K | None) → None¶

Called by SyncableDroppable when synced up to (earliest timestamp synced) is changed.

Parameters:: synced_up_to – new value of synced up to

abstract put(key: K, value: V) → None¶

Put given value to storage at given key.

This may block for a while.

Parameters:

key – key to store
value – value to store

SparseMatrix¶

class satella.coding.structures.SparseMatrix(matrix_data: List[List[T]] | None = None)¶

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|

append_row(y: Iterable[T]) → None¶

Append a row to the bottom of the matrix

Parameters:: y – iterable with consequent columns

clear() → None¶: Clear the contents of the sparse matrix

property columns: int¶: Return the amount of columns

delete_row(row_no: int) → None¶

Delete a row with specified number

Parameters:: row_no – number of the row to delete

classmethod from_iterable(y: Iterable[Iterable[T]])¶

Construct a sparse matrix given a row-first iterable. That iterable must return another iterable, that will yield values for given column.

Parameters:: y – an iterable describing the sparse matrix
Returns:: a sparse matrix object

get_neighbour_coordinates(col: int, row: int, include_diagonals: bool = True) → Iterator[Tuple[int, int]]¶: 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)

get_row(row_no: int) → List[T]¶

Return a single row of provided number.

The returned array has the same length as .columns

Parameters:: row_no – row number, numbered from 0

max() → T¶

Return maximum element.

None elements will be ignored.

min() → T¶

Return minimum element.

None elements will be ignored.

property rows: int¶: Return the amount of rows

shoot() → SparseMatrix¶: 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]]

Heaps¶

Heap¶

This essentially allows you to have a heap object that will pretty much behave like the heapq library.

class satella.coding.structures.Heap(from_list: Iterable[T] | None = None)¶

Sane heap as object - not like heapq.

Goes from lowest-to-highest (first popped is smallest). Standard Python comparision rules apply.

Not thread-safe

filter_map(filter_fun: Callable[[T], bool] | None = None, map_fun: Callable[[T], Any] | None = None)¶

Get only items that return True when condition(item) is True. Apply a: transform: item’ = item(condition) on

the rest. Maintain heap invariant.

iter_ascending() → Iterable[T]¶: Return an iterator returning all elements in this heap sorted ascending. State of the heap is not changed

iter_descending() → Iterable[T]¶

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.

pop() → T¶

Return smallest element of the heap.

Raises:: IndexError – on empty heap

pop_item(item: T) → T¶

Pop an item off the heap, maintaining the heap invariant

Raises:: ValueError – element not found

push(item: T) → None¶

Use it like:

>>> heap.push(3)

or:

>>> heap.push(4, myobject)

SetHeap¶

A heap with additional invariant that no two elements on the heap are the same. This is optimized for fast pushes() and membership checks.

class satella.coding.structures.SetHeap(from_list: Iterable[T] | None = None)¶

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

filter_map(filter_fun: Callable[[T], bool] | None = None, map_fun: Callable[[T], Any] | None = None)¶

Get only items that return True when condition(item) is True. Apply a: transform: item’ = item(condition) on

the rest. Maintain heap invariant.

pop() → T¶

Return smallest element of the heap.

Raises:: IndexError – on empty heap

push(item: T)¶

Use it like:

>>> heap.push(3)

or:

>>> heap.push(4, myobject)

TimeBasedHeap¶

Time-based heap is a good structure if you have many callbacks set to fire at a particular time in the future. It functions very like a normal Heap.

class satella.coding.structures.TimeBasedHeap(default_clock_source: ~typing.Callable[[], int | float] = <built-in function monotonic>)¶

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

get_timestamp(item: T) → int | float¶: Return the timestamp for given item

items() → Iterable[Tuple[int | float, T]]¶: Return an iterable, but WITHOUT timestamps (only items), in unspecified order

peek_closest() → Tuple[int | float, T]¶

Return the closest object to the execution deadline, but not discard it from the heap.

Raises:: IndexError – the heap is empty

pop_item(item: T) → Tuple[int | float, T]¶

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

pop_less_than(less: int | float | None = None) → Iterator[Tuple[int | float, T]]¶

Return all elements less (sharp inequality) than particular value.

This changes state of the heap

Parameters:: less – value to compare against. If left at default, it will be the default clock source specified at construction.
Returns:: an Iterator

pop_timestamp(timestamp: int | float) → T¶

Get first item with given timestamp, while maintaining the heap invariant

Raises:: ValueError – element not found

put(timestamp_or_value: T | int | float, value: T | None = None) → None¶

Put an item on heap.

Pass timestamp, item or just an item for default time

remove(item: T) → None¶: Remove all things equal to item

TimeBasedSetHeap¶

A combination of TimeBasedHeap and SetHeap:

class satella.coding.structures.TimeBasedSetHeap(default_clock_source: Callable[[], int | float] | None = None)¶

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

get_timestamp(item: T) → int | float¶

Return the timestamp for given item

Raises:: ValueError – item not found

items() → Iterator[T]¶: Return an iterator, but WITHOUT timestamps (only items), in unspecified order

pop() → Tuple[int | float, T]¶

Return smallest element of the heap.

Raises:: IndexError – on empty heap

pop_item(item: T) → Tuple[int | float, T]¶

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

pop_less_than(less: int | float | None = None) → Iterator[Tuple[int | float, T]]¶

Return all elements less (sharp inequality) than particular value.

This changes state of the heap

Parameters:: less – value to compare against
Returns:: a Generator

pop_timestamp(timestamp: int | float) → T¶

Pop an arbitary object (in case there’s two) item with given timestamp, while maintaining the heap invariant

Raises:: ValueError – element not found

push(item: Tuple[int | float, T]) → None¶

Use it like:

>>> heap.push(3)

or:

>>> heap.push(4, myobject)

put(timestamp_or_value: T | int | float, value: T | None = None) → None¶

Put an item on heap.

Pass timestamp, item or just an item for default time

remove(item: T) → None¶: Remove all things equal to item

Mixins¶

Mixins are classes whose constructor you do not need to invoke. They magically endow your class with chosen properties, often by overloading their specific special methods.

OnStrOnlyName¶

class satella.coding.structures.OnStrOnlyName¶

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'

HashableMixin¶

class satella.coding.structures.HashableMixin¶

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__

ComparableAndHashableByInt¶

class satella.coding.structures.ComparableAndHashableByInt¶: A mix-in. Provides comparison (lt, gt, ge, le, eq) and hashing by __int__ of this class.

OmniHashableMixin¶

If you need quick __hash__ and __eq__ operators from listed fields of the class.

class satella.coding.structures.OmniHashableMixin¶

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

ComparableAndHashableBy¶

class satella.coding.structures.ComparableAndHashableBy¶

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()

StrEqHashableMixin¶

A class that outfits your class with __eq__ and __hash__ based off the str() value of the class. So you got to define __str__ at the very least!

class satella.coding.structures.StrEqHashableMixin¶: A mix-in that outfits your class with an __eq__ and __hash__ operator that take their values from __str__ representation of your class.

ReprableMixin¶

If you need to provide a quick __repr__ for your classes:

class satella.coding.structures.ReprableMixin¶

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')")

ComparableEnum¶

class satella.coding.structures.ComparableEnum(value)¶

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

HashableIntEnum¶

An enum.IntEnum that can be __hash__ed

class satella.coding.structures.HashableIntEnum(value)¶: An enum.IntEnum that implements hashability, stemming from its values, as well as hashability

ComparableIntEnum¶

An enum.IntEnum that you can compare by it’s values

class satella.coding.structures.ComparableIntEnum(value)¶

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.

Immutable¶

Make your classes immutable. Normal assignment is only supported in the constructor, anywhere else it’s a TypeError.

Immutable inherits from abc.ABCMeta, so it’s safe to use abstract base classes here.

class Test(Immutable, metaclass=ABCMeta):

    attr: str = None

    def __init__(self):
        self.attr = 'value'

    def illegal_op(self):
        self.attr = 'test'  # this will TypeError

Singletons¶

Singleton¶

Makes the resulting object’s __init__() be called at most once, then caches the object and returns the same upon each instantiation.

satella.coding.structures.Singleton(cls)¶

Make a singleton out of decorated class.

Usage:

>>> @Singleton
>>> class MyClass:
>>>     ...

SingletonWithRegardsTo¶

Sometimes you just need an almost-singleton class, ie. class whose instance will depend on first n arguments. This function makes it easy:

satella.coding.structures.SingletonWithRegardsTo(num_args: int)¶

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

It will remember instances already created and return you a previously created instance, keying on the first n arguments.

There are also two functions to help you with managing your SingletonWithRegardsTo:

satella.coding.structures.get_instances_for_singleton(x) → List[Tuple]¶

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

satella.coding.structures.delete_singleton_for(x, *args) → None¶

Delete singleton for given arguments in a class decorated with SingletonWithRegardsTo

Parameters:

x – class decorated with SingletonWithRegardsTo
args – arguments used in the constructor

Dictionaries¶

DefaultDict¶

class satella.coding.structures.DefaultDict¶: A collections.defaultdict that does not store in itself empty values that it generates

CountingDict¶

class satella.coding.structures.CountingDict(set_to_count: Sequence = ())¶

A dictionary to quickly count the amount of elements in a set

Parameters:: set_to_count – a sequence, whose elements should be counted. The elements should support being keys in a dictionary.

count(elem, pieces: int = 1) → None¶

Add an instance of elem to the set

Parameters:

elem – instance to add
pieces – amount to add

LRU¶

class satella.coding.structures.LRU¶

A class to track least recently used objects.

get_item_to_evict() → T¶: Return a least recently used object

ExclusiveWritebackCache¶

class satella.coding.structures.ExclusiveWritebackCache(write_method: Callable[[K, V], None], read_method: Callable[[K], V], delete_method: Callable[[K], None] | None = None, executor: Executor | None = None, no_concurrent_executors: int | None = None, store_key_errors: bool = True)¶

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 – a blocking callable (key, value) that writes the value to underlying storage.
read_method – 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, 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 – an executor to execute the calls with. If None (default) is given, a ThreadPoolExecutor with 4 workers will be created
no_concurrent_executors – number of concurrent jobs that the executor is able to handle. This is used by sync()
store_key_errors – whether to remember KeyErrors raised by read_method

Use it as you would a normal dictionary.

DictObject¶

DictObject is an object constructed out of a dict, that allows it’s values to be obtained as getattr(), and not only getitem().

class satella.coding.structures.DictObject¶

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: Descriptor | Dict | None = None, **kwarg_schema) → bool¶

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 – schema to verify against
Returns:: whether is conformant

You can use the following function to recursively turn every dict into a DictObject

satella.coding.structures.apply_dict_object(v: Any | Dict) → DictObject | Any¶

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.

frozendict¶

A dictionary that can’t be modified. I didn’t import the one from _frozendict PyPI package, because it failed on Python 3.9. It is additionally hashable and __eq__-able

class satella.coding.structures.frozendict¶: 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.

DictionaryView¶

class satella.coding.structures.DictionaryView(master_dict: Dict[K, V], *rest_of_dicts: Dict[K, V], propagate_deletes: bool = True, assign_to_same_dict: bool = True)¶

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 – First dictionary to look up. Entries made via __setitem__ will be put here.
rest_of_dicts – Remaining dictionaries
propagate_deletes – 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 – 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.

CacheDict¶

class satella.coding.structures.CacheDict(stale_interval: float, expiration_interval: float, value_getter: ~typing.Callable[[~satella.coding.typing.K], ~satella.coding.typing.V], value_getter_executor: ~concurrent.futures._base.Executor | None = None, cache_failures_interval: float | None = None, time_getter: ~typing.Callable[[], float] = <built-in function monotonic>, default_value_factory: ~typing.Callable[[], ~satella.coding.typing.V] | None = None)¶

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 – 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
expiration_interval – 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
value_getter – 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 – 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 – 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.
time_getter – a routine used to get current time in seconds
default_value_factory – 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: K, value: V, timestamp: float | None = None)¶: Feed this data into the cache

get_value_block(key: K) → V¶

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

has_info_about(key: K) → bool¶: Is provided key cached, or failure about it is cached?

invalidate(key: K) → None¶

Remove all information about given key from the cache

Syntactic sugar for:

>>> try:
>>>   del self[key]
>>> except KeyError:
>>>   pass

schedule_a_fetch(key: K) → Future¶

Schedule a value refresh for given key

Parameters:: key – key to schedule the refresh for
Returns:: future that was queued to ask for given key

LRUCacheDict¶

class satella.coding.structures.LRUCacheDict(*args, max_size: int = 100, **kwargs)¶

A dictionary that you can use as a cache with a maximum size, items evicted by LRU policy.

Parameters:: max_size – maximum size

feed(key: K, value: V, timestamp: float | None = None)¶: Feed this data into the cache

get_value_block(key: K) → V¶

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

invalidate(key: K) → None¶

Remove all information about given key from the cache

Syntactic sugar for:

>>> try:
>>>   del self[key]
>>> except KeyError:
>>>   pass

make_room() → None¶: Assure that there’s place for at least one element

SelfCleaningDefaultDict¶

class satella.coding.structures.SelfCleaningDefaultDict(default_factory: Callable[[], V], background_maintenance: bool = True, *args, **kwargs)¶

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 and satella.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 – a callable/0 that will return an object if it doesn’t already exist
background_maintenance – 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.

ExpiringEntryDict¶

class satella.coding.structures.ExpiringEntryDict(expiration_timeout: float, *args, time_getter: ~typing.Callable[[], float] = <built-in function monotonic>, external_cleanup: bool = False, **kwargs)¶

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 – number of seconds after which entries will expire
time_getter – a callable/0 that returns the current timestamp
external_cleanup – 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.

cleanup() → None¶: Remove entries that are later than given time

get_timestamp(key: K) → float¶

Return the timestamp at which given key was inserted in the dict

Raises:: KeyError – key not found in the dictionary

TwoWayDictionary¶

class satella.coding.structures.TwoWayDictionary(data=None, _is_reverse: bool = False)¶

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
Raises:: ValueError – on being given data from which it is impossible to construct a reverse mapping (ie. same value appears at least twice)

done() → None¶

Called when the user is done using given TwoWayDictionary.

Internally this will break the reference cycle, and enable Python GC to collect the objects.

property reverse: MutableMapping[V, K]¶: Return a reverse mapping. Reverse mapping is updated as soon as an operation is done.

DirtyDict¶

A dictionary that has also a flag called dirty that says if it’s been last modified since that flag was cleared.

The flag is initially (after the dict has been created) set to False.

class satella.coding.structures.DirtyDict(*args, **kwargs)¶

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() → None¶: Clears the dirty flag

copy_and_clear_dirty() → Dict[K, V]¶

Returns a copy of this data and sets dirty to False

Returns:: a plain, normal Python dictionary is returned

swap_and_clear_dirty() → Dict[K, V]¶

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

KeyAwareDefaultDict¶

class satella.coding.structures.KeyAwareDefaultDict(factory_function: Callable[[K], V], *args, **kwargs)¶

A defaultdict whose factory function accepts the key to provide a default value for the key

Parameters:: factory_function – a callable that accepts a single argument, a key, for which it is to provide a default value

Other structures¶

typednamedtuple¶

It’s a named tuple, but it has typed fields. You will get a TypeError if you try to assign something else there.

satella.coding.structures.typednamedtuple(cls_name: str, *arg_name_type: type) → Type[Tuple]¶

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

HashableWrapper¶

class satella.coding.structures.HashableWrapper(object_to_wrap: T, wrap_operations: bool = False)¶

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)

Ranking¶

class satella.coding.structures.Ranking(items: List[T], key: Callable[[T], int])¶

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

add(item: T) → None¶: Add a single element to the ranking and _recalculate it

get_position_of(item: T) → int¶

Return the position in the ranking of element item

Parameters:: item – element to return the position for
Returns:: position
Raises:: ValueError – this element is not in the ranking

get_sorted() → Iterator[T]¶: Return all the elements sorted from the least key value to the highest key value

remove(item: T) → None¶: Remove a single element from the ranking and _recalculate it

SortedList¶

class satella.coding.structures.SortedList(items: ~typing.Iterable[~satella.coding.typing.T] = (), key: ~typing.Callable[[~satella.coding.typing.T], int] = <function SortedList.<lambda>>)¶

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 – items to construct the list with
key – a callable[T]->int that builds the key of the sort

add(other: T) → int¶

Add an element. Returns the index at which it was inserted.

Parameters:: other – element to insert
Returns:: index that the entry is available now at

extend(elements: Iterable[T])¶: Adds multiple elements to this list

index(other: T) → int¶: Return index at which given value has been placed

pop() → T¶: Return the highest element, removing it from the list

popleft() → T¶: Return the smallest element, removing it from the list

remove(other: T) → None¶

Remove an element from the list

Parameters:: other – element to remove
Raises:: ValueError – element not in list

SliceableDeque¶

class satella.coding.structures.SliceableDeque(*args, **kwargs)¶

A collections.deque that supports slicing.

Just note that it will return a p_gen upon being sliced!

insert(i: int, item: T)¶: S.insert(index, value) – insert value before index

Proxy¶

class satella.coding.structures.Proxy(object_to_wrap: T, wrap_operations: bool = False)¶

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 – object to wrap
wrap_operations – whether results of operations returning something else should be also proxied.

Subqueue¶

class satella.coding.structures.Subqueue¶

Or a named queue is a collection of thread safe queues identified by name

assert_queue(queue_name: str) → None¶: Assure that we have a queue with a particular name in the dictionary

get(queue_name: str, block=True, timeout=None) → T¶: Same semantics at queue.Queue.get

get_any() → Tuple[str, T]¶

Return any message with name of the queue.

This assumes block=False

Returns:: a tuple of (queue_name, object)
Raises:: queue.Empty()

put(queue_name: str, obj: T) → None¶: Same semantics as queue.Queue.put

qsize() → int¶: Calculate the total of entries

Closeable¶

class satella.coding.Closeable¶

A class that needs to clean up its own resources.

It’s destructor calls .close(). Use like this:

>>> class MyClose(Closeable):
>>>     def close(self):
>>>         if super().close():
>>>             .. clean up ..

Can be also used as a context manager, with close() called upon __exit__.

Warning

You should extend both __init__ and close(). Invoke __init__() at the end of your class constructor, this will prevent the destructor from closing on half-initialized classes.

Objects before initialization (calling of this constructor) are considered closed. Checking if they are closed will emit a warning.

close() → bool¶

Check if the resource needs cleanup, and clean up this resource.

Use like this:

>>> class MyClose(Closeable):
>>>     def close(self):
>>>         if super().close():
>>>             .. clean up ..

Returns:: whether the cleanup should proceed
Raises:: RuntimeError – the constructor was not invoked

property closed: bool¶

Returns:: whether this object is closed