fleche.storage.base

Attributes

logger

Exceptions

SaveError	Common base class for all non-exit exceptions.
AmbiguousDigestError	Inappropriate argument value (of correct type).

Classes

StorageBase	Shared functionality between value and call storages.
Storage	Abstract base class for defining storage mechanisms.
Digested	Helper class that provides a standard way to create an ABC using
DigestedIterable	Helper class that provides a standard way to create an ABC using
DigestedDict	Helper class that provides a standard way to create an ABC using
DestructuringMixin	Mixin that recursively destructures collections on save/load.
_DelegatingStorage	Storage that delegates all operations to a wrapped storage instance.
DestructuringStorage	Storage wrapper that recursively destructures collections.
CallStorage	Special storage for saving Call instances.
CallStorageAdapter	Implement a CallStorage from a generic Storage.

Module Contents

fleche.storage.base.logger

exception fleche.storage.base.SaveError

Bases: Exception

Common base class for all non-exit exceptions.

exception fleche.storage.base.AmbiguousDigestError

Bases: ValueError

Inappropriate argument value (of correct type).

class fleche.storage.base.StorageBase

Bases: abc.ABC

Shared functionality between value and call storages.

abstractmethod list() → Iterable[fleche.digest.Digest]

evict(key: fleche.digest.Digest | str) → None

Removes the entry corresponding to the key from the storage.

abstractmethod _evict(key: fleche.digest.Digest) → None

expand(key: fleche.digest.Digest | str) → fleche.digest.Digest

Expands a short-hand digest to the full length one.

shrink(key: fleche.digest.Digest | str) → fleche.digest.Digest

Find the shortest substring that is still an unambigious reference to the same value.

contains(key: fleche.digest.Digest | str) → bool

abstractmethod _contains(key: fleche.digest.Digest) → bool

class fleche.storage.base.Storage

Bases: StorageBase

Abstract base class for defining storage mechanisms.

save(value: Any, key: fleche.digest.Digest | None = None) → fleche.digest.Digest

abstractmethod _save(value: Any, key: fleche.digest.Digest) → fleche.digest.Digest

load(key: fleche.digest.Digest | str) → Any

abstractmethod _load(key: fleche.digest.Digest) → Any

_contains(key: fleche.digest.Digest) → bool

class fleche.storage.base.Digested

Bases: abc.ABC

Helper class that provides a standard way to create an ABC using inheritance.

abstractmethod underlying()

Return plain underlying value, ie. list/dict/etc of nested values or their partial digests

__digest__()

abstractmethod mend(storage: DestructuringMixin)

classmethod sunder(intern: Callable[[Any], tuple[Any, int | float]], value: Any)

Abstractmethod:

class fleche.storage.base.DigestedIterable

Bases: Digested

Helper class that provides a standard way to create an ABC using inheritance.

items: list | tuple

underlying()

Return plain underlying value, ie. list/dict/etc of nested values or their partial digests

mend(storage: DestructuringMixin) → list | tuple

classmethod sunder(intern: Callable[[Any], tuple[Any, int | float]], value: list | tuple)

class fleche.storage.base.DigestedDict

Bases: Digested

Helper class that provides a standard way to create an ABC using inheritance.

items: dict

underlying()

Return plain underlying value, ie. list/dict/etc of nested values or their partial digests

mend(storage: DestructuringMixin) → dict

classmethod sunder(intern: Callable[[Any], tuple[Any, int | float]], value: dict)

class fleche.storage.base.DestructuringMixin

Bases: Storage

Mixin that recursively destructures collections on save/load.

Place before a concrete Storage in the MRO to add destructuring behavior. Lists, tuples, and dicts are broken apart so each element is stored independently; on load the original structure is reassembled.

Example:

class DestructuringMemory(DestructuringMixin, Memory):
    pass

# ``storage`` here is the backing dict required by ``Memory``, not a
# Storage instance.
dm = DestructuringMemory(storage={})
key = dm.save([1, [2, 3]])
assert dm.load(key) == [1, [2, 3]]

remaining_depth: int = 0

_intern_rec(value: Any) → tuple[Any, int | float]

Post-order traversal: recurse to leaves, decide inline-vs-store on the way back up.

Returns (result, depth) where result is the plain value when depth < remaining_depth (the element is inlined in its parent’s Digested wrapper) or a Digest when the element was written to storage separately. Every node in the structure is visited exactly once (O(n)), unlike a separate depth-counting pass.

_save(value: Any, key: fleche.digest.Digest) → fleche.digest.Digest

_load(key: fleche.digest.Digest | Any) → Any

class fleche.storage.base._DelegatingStorage

Bases: Storage

Storage that delegates all operations to a wrapped storage instance.

storage: Storage

_save(value: Any, key: fleche.digest.Digest) → fleche.digest.Digest

_load(key: fleche.digest.Digest) → Any

_contains(key: fleche.digest.Digest) → bool

_evict(key: fleche.digest.Digest) → None

list() → Iterable[fleche.digest.Digest]

class fleche.storage.base.DestructuringStorage

Bases: DestructuringMixin, _DelegatingStorage

Storage wrapper that recursively destructures collections.

This is a convenience class combining DestructuringMixin with delegation to a wrapped storage. Prefer using DestructuringMixin directly as a mixin with a concrete storage class when possible.

remaining_depth: int = 0

__post_init__()

class fleche.storage.base.CallStorage

Bases: StorageBase

Special storage for saving Call instances.

save(call: fleche.call.Call) → fleche.digest.Digest

abstractmethod _save(call: fleche.call.Call) → fleche.digest.Digest

load(key: str | fleche.digest.Digest) → fleche.call.Call

abstractmethod _load(key: fleche.digest.Digest) → fleche.call.Call

transform(func: Callable[[fleche.call.Call], fleche.call.Call] | None = None) → None

Applies a transformation function to all Call objects in the storage.

Parameters:

func (Callable[[Call], Call] | None) – A function that takes a Call and returns a transformed Call. If None, the identity function is used (useful for re-calculating keys).

query(template: fleche.call.QueryCall) → Iterable[fleche.call.Call]

Find cached calls that ‘match’ the template.

Returns all calls where the given arguments, results or metadata match exactly the stored ones. Values may be given either as they are or as Digest.

Parameters:

template (Call) – specification for calls to return; use None as wildcard.

Returns:

an iterable over all matching call objects

Return type:

Iterable[Call]

_contains(key: fleche.digest.Digest) → bool

class fleche.storage.base.CallStorageAdapter

Bases: CallStorage

Implement a CallStorage from a generic Storage.

storage: Storage

_save(call: fleche.call.Call) → fleche.digest.Digest

_load(key: fleche.digest.Digest) → fleche.call.Call

_contains(key: fleche.digest.Digest) → bool

_evict(key: fleche.digest.Digest) → None

list() → Iterable[fleche.digest.Digest]