ObjectsLocalStorage.py

from shelve import DbfilenameShelf
from typing import Any
from typing import Type
from typing import TypeVar

T = TypeVar('T')
__all__ = ['ObjectsLocalStorage']


class ObjectsLocalStorage(DbfilenameShelf):
    def __getitem__(self, key: Type[T]) -> T:
        key = key.__name__
        try:
            value = self.cache[key]
        except KeyError:
            f = BytesIO(self.dict[key.encode(self.keyencoding)])
            value = Unpickler(f).load()
            if self.writeback:
                self.cache[key] = value
        return value

    def __setitem__(self, key: Type[T], value):
        key = key.__name__
        if self.writeback:
            self.cache[key] = value
        f = BytesIO()
        p = Pickler(f, self._protocol)
        p.dump(value)
        self.dict[key.encode(self.keyencoding)] = f.getvalue()

    def sync(self):
        if self.writeback and self.cache:
            self.writeback = False
            for key, entry in self.cache.items():
                super().__setitem__(key, entry)
            self.writeback = True
            self.cache = {}
        if hasattr(self.dict, 'sync'):
            self.dict.sync()

    def get(self, key: Type[T], default=None) -> T:
        key = key.__name__
        if key in self.dict:
            return super().__getitem__(key)
        return default