from dataclasses import dataclass
from typing import Any, Iterable
from .base import ValueMixin, CallMixin, StorageBackend
from .destructuring import DestructuringMixin
from .thread_safe import PerKeyLockMixin
from ..digest import Digest
from copy import deepcopy
@dataclass(frozen=True)
[docs]
class MemoryBackend(StorageBackend):
"""
A concrete implementation of Storage that stores values in an in-memory dictionary.
"""
[docs]
storage: dict[Digest, Any]
# Each storage instance is its own world: hash on identity so that frozen
# dataclass subclasses can serve as WeakKeyDictionary keys in PerKeyLockMixin
# without the indigestible dict field causing a TypeError.
[docs]
__hash__ = object.__hash__
[docs]
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
if "__hash__" not in cls.__dict__:
raise TypeError(
f"{cls.__qualname__} subclasses MemoryBackend without defining "
"`__hash__ = object.__hash__`. The inherited `storage: dict` field is "
"unhashable, and `@dataclass(frozen=True)` regenerates `__hash__` on "
"every subclass without carrying the override through. "
"Add `__hash__ = object.__hash__` to the class body."
)
[docs]
def list(self) -> Iterable[Digest]:
return tuple(self.storage.keys())
[docs]
def put(self, value: Any, key: Digest) -> Digest:
self.storage[key] = deepcopy(value)
return key
[docs]
def get(self, key: Digest) -> Any:
return deepcopy(self.storage[key])
[docs]
def _contains(self, key: Digest) -> bool:
return key in self.storage
[docs]
def _evict(self, key: Digest) -> None:
self.storage.pop(key, None)
@dataclass(frozen=True)
[docs]
class ValueMemory(PerKeyLockMixin, DestructuringMixin, ValueMixin, MemoryBackend):
[docs]
__hash__ = object.__hash__
@dataclass(frozen=True)
[docs]
class CallMemory(PerKeyLockMixin, CallMixin, MemoryBackend):
[docs]
__hash__ = object.__hash__