Source code for errbot.storage

import logging
import types
from collections.abc import MutableMapping
from contextlib import contextmanager

log = logging.getLogger(__name__)


[docs]class StoreException(Exception): pass
[docs]class StoreAlreadyOpenError(StoreException): pass
[docs]class StoreNotOpenError(StoreException): pass
[docs]class StoreMixin(MutableMapping): """ This class handle the basic needs of bot plugins and core like loading, unloading and creating a storage """
[docs] def __init__(self): self._store = None self.namespace = None
[docs] def open_storage(self, storage_plugin, namespace): if self.is_open_storage(): raise StoreAlreadyOpenError("Storage appears to be opened already") log.debug("Opening storage '%s'", namespace) self._store = storage_plugin.open(namespace) self.namespace = namespace
[docs] def close_storage(self): if not self.is_open_storage(): raise StoreNotOpenError(f"Storage does not appear to have been opened yet") self._store.close() self._store = None log.debug("Closed storage '%s'", self.namespace)
[docs] def is_open_storage(self): has_store_key = hasattr(self, "_store") if has_store_key and self._store: return True elif not has_store_key or self._store is None: return False else: return False
# those are the minimal things to behave like a dictionary with the UserDict.DictMixin def __getitem__(self, key): return self._store.get(key)
[docs] @contextmanager def mutable(self, key, default=None): try: obj = self._store.get(key) except KeyError: obj = default yield obj # implements autosave for a plugin persistent entry # with self['foo'] as f: # f[4] = 2 # saves the entry ! self._store.set(key, obj)
def __setitem__(self, key, item): return self._store.set(key, item) def __delitem__(self, key): return self._store.remove(key)
[docs] def keys(self): return self._store.keys()
def __len__(self): return self._store.len() def __iter__(self): for i in self._store.keys(): yield i def __contains__(self, x): try: self._store.get(x) return True except KeyError: return False # compatibility with with def __enter__(self): return self def __exit__(self, type, value, traceback): self.close_storage()