import logging from typing import List from common.database.api.Exceptions import WrongDatabaseEngine, MutexException from common.database.api.context.Context import Context from common.database.api.context.Keys import KEY_CONTEXTS from common.database.api.entity.EntityCollection import EntityCollection from common.database.engines._DatabaseEngine import _DatabaseEngine LOGGER = logging.getLogger(__name__) class Database: def __init__(self, database_engine : _DatabaseEngine): if not isinstance(database_engine, _DatabaseEngine): raise WrongDatabaseEngine('database_engine must inherit from _DatabaseEngine') self._database_engine = database_engine self._acquired = False self._owner_key = None self._contexts = EntityCollection(self, KEY_CONTEXTS) @property def database_engine(self) -> _DatabaseEngine: return self._database_engine def __enter__(self) -> '_DatabaseEngine': self._acquired, self._owner_key = self._database_engine.lock() if not self._acquired: raise MutexException('Unable to acquire database lock') return self def __exit__(self, exc_type, exc_val, exc_tb): self._database_engine.unlock(self._owner_key) def clear_all(self, keep_keys=set()): LOGGER.info('Cleaning up...') keys = self._database_engine.keys() LOGGER.info(' keys before = {}'.format(str(keys))) for key in keys: if(key in keep_keys): continue self._database_engine.delete(key) LOGGER.info(' keys after = {}'.format(str(self._database_engine.keys()))) def dump(self) -> List[str]: entries = self._database_engine.dump() entries.sort() return ['[{:>4s}] {:100s} :: {}'.format(k_type, k_name, k_value) for k_name,k_type,k_value in entries] @property def contexts(self) -> EntityCollection: return self._contexts def context(self, context_uuid : str) -> Context: return Context(context_uuid, self._database_engine)