Source code for caspia.toolbox.storage.base

import collections.abc
from abc import ABCMeta, abstractmethod


[docs]class Storage(collections.MutableMapping, metaclass=ABCMeta): def __init__(self): self._store = None @abstractmethod def _get_store(self): """Return underlying store. Must return object behaving as dict with .close() method. """ def __enter__(self) -> collections.abc.MutableMapping: if self._store: raise RuntimeError('%s is already opened' % self) self._store = self._get_store() return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: self._store.close() self._store = None def _ensure_opened(self): if self._store is None: raise RuntimeError('%s is not opened. Use `with` statement' % self) def __getitem__(self, key): self._ensure_opened() return self._store.__getitem__(key) def __setitem__(self, key, value): self._ensure_opened() return self._store.__setitem__(key, value) def __delitem__(self, key): self._ensure_opened() return self._store.__delitem__(key) def __iter__(self): self._ensure_opened() return self._store.__iter__() def __len__(self): self._ensure_opened() return self._store.__len__()