Source code for errbot.storage.shelf

import logging
import os
import shelve
import shutil
from typing import Any

from errbot.storage.base import StorageBase, StoragePluginBase

log = logging.getLogger("errbot.storage.shelf")


[docs]class ShelfStorage(StorageBase):
[docs] def __init__(self, path): log.debug("Open shelf storage %s", path) self.shelf = shelve.DbfilenameShelf(path, protocol=2)
[docs] def get(self, key: str) -> Any: return self.shelf[key]
[docs] def remove(self, key: str): if key not in self.shelf: raise KeyError(f"{key} doesn't exist.") del self.shelf[key]
[docs] def set(self, key: str, value: Any) -> None: self.shelf[key] = value
[docs] def len(self): return len(self.shelf)
[docs] def keys(self): return self.shelf.keys()
[docs] def close(self) -> None: self.shelf.close() self.shelf = None
[docs]class ShelfStoragePlugin(StoragePluginBase):
[docs] def __init__(self, bot_config): super().__init__(bot_config) if "basedir" not in self._storage_config: self._storage_config["basedir"] = bot_config.BOT_DATA_DIR
[docs] def open(self, namespace: str) -> StorageBase: config = self._storage_config # Hack to port move old DBs to the new location. new_spot = os.path.join(config["basedir"], namespace + ".db") old_spot = os.path.join(config["basedir"], "plugins", namespace + ".db") if os.path.isfile(old_spot): if os.path.isfile(new_spot): log.warning( "You have an old v3 DB at %s and a duplicate new one at %s.", old_spot, new_spot, ) log.warning( "You need to either remove the old one or move it in place of the new one manually." ) else: log.info("Moving your old v3 DB from %s to %s.", old_spot, new_spot) shutil.move(old_spot, new_spot) return ShelfStorage(new_spot)