Module: storage

class omemo.storage.Just(value)[source]

Bases: Maybe[ValueTypeT]

A Maybe that does hold a value.

Parameters:: value (ValueTypeT)

__init__(value)[source]

Initialize a Just, representing a Maybe that holds a value.

Parameters:: value (TypeVar(ValueTypeT)) – The value to store in this Just.
Return type:: None

property is_just: bool: Returns: Whether this is a Just.

property is_nothing: bool: Returns: Whether this is a Nothing.

from_just()[source]

Return type:: TypeVar(ValueTypeT)
Returns:: The value if this is a Just.
Raises:: NothingException – if this is a Nothing.

maybe(default)[source]

Parameters:: default (TypeVar(DefaultTypeT)) – The value to return if this is in instance of Nothing.
Return type:: TypeVar(ValueTypeT)
Returns:: The value if this is a Just, or the default value if this is a Nothing. The default is returned by reference in that case.

fmap(function)[source]

Apply a mapping function.

Parameters:: function (Callable[[TypeVar(ValueTypeT)], TypeVar(MappedValueTypeT)]) – The mapping function.
Return type:: Just[TypeVar(MappedValueTypeT)]
Returns:: A new Just containing the mapped value if this is a Just. A new Nothing if this is a Nothing.

class omemo.storage.Maybe[source]

Bases: ABC, Generic[ValueTypeT]

typing’s Optional[A] is just an alias for Union[None, A], which means if A is a union itself that allows None, the Optional[A] doesn’t add anything. E.g. Optional[Optional[X]] = Optional[X] is true for any type X. This Maybe class actually differenciates whether a value is set or not.

All incoming and outgoing values or cloned using copy.deepcopy(), such that values stored in a Maybe instance are not affected by outside application logic.

abstract property is_just: bool: Returns: Whether this is a Just.

abstract property is_nothing: bool: Returns: Whether this is a Nothing.

abstractmethod from_just()[source]

Return type:: TypeVar(ValueTypeT)
Returns:: The value if this is a Just.
Raises:: NothingException – if this is a Nothing.

abstractmethod maybe(default)[source]

Parameters:: default (TypeVar(DefaultTypeT)) – The value to return if this is in instance of Nothing.
Return type:: TypeVar(ValueTypeT) | TypeVar(DefaultTypeT)
Returns:: The value if this is a Just, or the default value if this is a Nothing. The default is returned by reference in that case.

abstractmethod fmap(function)[source]

Apply a mapping function.

Parameters:: function (Callable[[TypeVar(ValueTypeT)], TypeVar(MappedValueTypeT)]) – The mapping function.
Return type:: Maybe[TypeVar(MappedValueTypeT)]
Returns:: A new Just containing the mapped value if this is a Just. A new Nothing if this is a Nothing.

class omemo.storage.Nothing[source]

Bases: Maybe[ValueTypeT]

A Maybe that does not hold a value.

__init__()[source]

Initialize a Nothing, representing an empty Maybe.

Return type:: None

property is_just: bool: Returns: Whether this is a Just.

property is_nothing: bool: Returns: Whether this is a Nothing.

from_just()[source]

Return type:: TypeVar(ValueTypeT)
Returns:: The value if this is a Just.
Raises:: NothingException – if this is a Nothing.

maybe(default)[source]

Parameters:: default (TypeVar(DefaultTypeT)) – The value to return if this is in instance of Nothing.
Return type:: TypeVar(DefaultTypeT)
Returns:: The value if this is a Just, or the default value if this is a Nothing. The default is returned by reference in that case.

fmap(function)[source]

Apply a mapping function.

Parameters:: function (Callable[[TypeVar(ValueTypeT)], TypeVar(MappedValueTypeT)]) – The mapping function.
Return type:: Nothing[TypeVar(MappedValueTypeT)]
Returns:: A new Just containing the mapped value if this is a Just. A new Nothing if this is a Nothing.

exception omemo.storage.NothingException[source]

Bases: Exception

Raised by Maybe.from_just(), in case the Maybe is a Nothing.

class omemo.storage.Storage(disable_cache=False)[source]

Bases: ABC

A simple key/value storage class with optional caching (on by default). Keys can be any Python string, values any JSON-serializable structure.

Warning

Writing (and deletion) operations must be performed right away, before returning from the method. Such operations must not be cached or otherwise deferred.

Warning

All parameters must be treated as immutable unless explicitly noted otherwise.

Note

The Maybe type performs the additional job of cloning stored and returned values, which essential to decouple the cached values from the application logic.

Parameters:: disable_cache (bool)

__init__(disable_cache=False)[source]

Configure caching behaviour of the storage.

Parameters:: disable_cache (bool) – Whether to disable the cache, which is on by default. Use this parameter if your storage implementation handles caching itself, to avoid pointless double caching.

abstractmethod async _load(key)[source]

Load a value.

Parameters:: key (str) – The key identifying the value.
Return type:: Maybe[JSONType]
Returns:: The loaded value, if it exists.
Raises:: StorageException – if any kind of storage operation failed. Feel free to raise a subclass instead.

abstractmethod async _store(key, value)[source]

Store a value.

Parameters:

key (str) – The key identifying the value.
value (JSONType) – The value to store under the given key.

Raises:

StorageException – if any kind of storage operation failed. Feel free to raise a subclass instead.

Return type:

None

abstractmethod async _delete(key)[source]

Delete a value, if it exists.

Parameters:: key (str) – The key identifying the value to delete.
Raises:: StorageException – if any kind of storage operation failed. Feel free to raise a subclass instead. Do not raise if the key doesn’t exist.
Return type:: None

async load(key)[source]

Load a value.

Parameters:: key (str) – The key identifying the value.
Return type:: Maybe[JSONType]
Returns:: The loaded value, if it exists.
Raises:: StorageException – if any kind of storage operation failed. Forwarded from _load().

async store(key, value)[source]

Store a value.

Parameters:

key (str) – The key identifying the value.
value (JSONType) – The value to store under the given key.

Raises:

StorageException – if any kind of storage operation failed. Forwarded from _store().

Return type:

None

async delete(key)[source]

Delete a value, if it exists.

Parameters:: key (str) – The key identifying the value to delete.
Raises:: StorageException – if any kind of storage operation failed. Does not raise if the key doesn’t exist. Forwarded from _delete().
Return type:: None

async store_bytes(key, value)[source]

Variation of store() for storing specifically bytes values.

Parameters:

key (str) – The key identifying the value.
value (bytes) – The value to store under the given key.

Raises:

StorageException – if any kind of storage operation failed. Forwarded from _store().

Return type:

None

async load_primitive(key, primitive)[source]

Variation of load() for loading specifically primitive values.

Parameters:

key (str) – The key identifying the value.
primitive (Type[TypeVar(PrimitiveTypeT, None, float, int, str, bool)]) – The primitive type of the value.

Return type:

Maybe[TypeVar(PrimitiveTypeT, None, float, int, str, bool)]

Returns:

The loaded and type-checked value, if it exists.

Raises:

StorageException – if any kind of storage operation failed. Forwarded from _load().

async load_bytes(key)[source]

Variation of load() for loading specifically bytes values.

Parameters:: key (str) – The key identifying the value.
Return type:: Maybe[bytes]
Returns:: The loaded and type-checked value, if it exists.
Raises:: StorageException – if any kind of storage operation failed. Forwarded from _load().

async load_optional(key, primitive)[source]

Variation of load() for loading specifically optional primitive values.

Parameters:

key (str) – The key identifying the value.
primitive (Type[TypeVar(PrimitiveTypeT, None, float, int, str, bool)]) – The primitive type of the optional value.

Return type:

Maybe[TypeVar(PrimitiveTypeT, None, float, int, str, bool) | None]

Returns:

The loaded and type-checked value, if it exists.

Raises:

StorageException – if any kind of storage operation failed. Forwarded from _load().

async load_list(key, primitive)[source]

Variation of load() for loading specifically lists of primitive values.

Parameters:

key (str) – The key identifying the value.
primitive (Type[TypeVar(PrimitiveTypeT, None, float, int, str, bool)]) – The primitive type of the list elements.

Return type:

Maybe[List[TypeVar(PrimitiveTypeT, None, float, int, str, bool)]]

Returns:

The loaded and type-checked value, if it exists.

Raises:

StorageException – if any kind of storage operation failed. Forwarded from _load().

async load_dict(key, primitive)[source]

Variation of load() for loading specifically dictionaries of primitive values.

Parameters:

key (str) – The key identifying the value.
primitive (Type[TypeVar(PrimitiveTypeT, None, float, int, str, bool)]) – The primitive type of the dictionary values.

Return type:

Maybe[Dict[str, TypeVar(PrimitiveTypeT, None, float, int, str, bool)]]

Returns:

The loaded and type-checked value, if it exists.

Raises:

StorageException – if any kind of storage operation failed. Forwarded from _load().

exception omemo.storage.StorageException[source]

Bases: OMEMOException

Parent type for all exceptions specifically raised by methods of Storage.

class omemo.storage.ValueTypeT: alias of TypeVar(‘ValueTypeT’)