266 lines
9.4 KiB
Python
266 lines
9.4 KiB
Python
import logging
|
|
from datetime import datetime
|
|
|
|
from constants import NOT_LOGGED, NO_SESSION
|
|
from core.dbengine import DbEngine, TAG_PARENT, TAG_USER, TAG_DATE, DbException
|
|
from core.settings_objects import *
|
|
|
|
load_settings_obj() # needed to make sure that the import of core is not removed
|
|
|
|
FAKE_USER_ID = "FakeUserId"
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class NoDefaultCls:
|
|
"""
|
|
No Default value class
|
|
Return when no default value is provided.
|
|
"""
|
|
pass
|
|
|
|
|
|
NoDefault = NoDefaultCls()
|
|
|
|
|
|
class MemoryDbEngine:
|
|
"""
|
|
Keeps everything in memory
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.db = {}
|
|
|
|
def init_db(self, entry, key, obj):
|
|
self.db[entry] = {key: obj}
|
|
|
|
def save(self, user_id: str, user_email: str, entry: str, obj: object) -> bool:
|
|
if user_id in self.db:
|
|
self.db[user_id][entry] = obj
|
|
self.db[user_id][TAG_PARENT] = [] # not used
|
|
self.db[user_id][TAG_USER] = user_email
|
|
self.db[user_id][TAG_DATE] = datetime.now().strftime('%Y%m%d %H:%M:%S %z')
|
|
else:
|
|
self.db[user_id] = {
|
|
entry: obj,
|
|
TAG_PARENT: [], # not set
|
|
TAG_USER: user_email,
|
|
TAG_DATE: datetime.now().strftime('%Y%m%d %H:%M:%S %z')
|
|
}
|
|
return True
|
|
|
|
def load(self, user_id: str, entry: str, digest: str = None):
|
|
try:
|
|
return self.db[user_id][entry]
|
|
except KeyError as e:
|
|
raise DbException(e)
|
|
|
|
def get(self, user_id: str, entry: str, key: str | None = None, digest=None):
|
|
"""
|
|
Retrieve a specific value or the entire object from stored data based on the
|
|
provided user ID and entry name. Optionally, a key can be specified to extract
|
|
a particular value from the loaded object.
|
|
|
|
:param user_id: The unique identifier of the user associated with the data
|
|
being retrieved.
|
|
:param entry: The name of the entry under the user's data storage.
|
|
:param key: Optional; Specific key for retrieving a particular value from the
|
|
entry. If not provided, the entire object is returned.
|
|
:param digest: Optional; Used to specify a digest or additional parameter,
|
|
but its function should be inferred from its use, as it is not directly
|
|
handled in this method.
|
|
:return: The value corresponding to the specified key within the user's entry,
|
|
or the entire entry object if no key is specified.
|
|
"""
|
|
obj = self.load(user_id, entry)
|
|
try:
|
|
return obj[key] if key else obj
|
|
except KeyError as e:
|
|
raise DbException(e)
|
|
|
|
def put(self, user_id: str, user_email: str, entry, key: str, value: object):
|
|
obj = self.load(user_id, entry)
|
|
obj[key] = value
|
|
|
|
def put_many(self, user_id: str, user_email: str, entry, items):
|
|
obj = self.load(user_id, entry)
|
|
obj.update(items)
|
|
|
|
def exists(self, user_id: str, entry: str):
|
|
return user_id in self.db and entry in self.db[user_id]
|
|
|
|
|
|
class SettingsManager:
|
|
def __init__(self, engine=None):
|
|
self._db_engine = engine or DbEngine()
|
|
|
|
def save(self, session: dict, entry: str, obj: object):
|
|
user_id, user_email = self._get_user(session)
|
|
return self._db_engine.save(user_id, user_email, entry, obj)
|
|
|
|
def load(self, session: dict, entry: str, digest=None, default=NoDefault):
|
|
user_id, _ = self._get_user(session)
|
|
try:
|
|
return self._db_engine.load(user_id, entry, digest)
|
|
except DbException:
|
|
return default
|
|
|
|
def put(self, session: dict, entry: str, key: str, value: object):
|
|
user_id, user_email = self._get_user(session)
|
|
return self._db_engine.put(user_id, user_email, entry, key, value)
|
|
|
|
def put_many(self, session: dict, entry: str, items: list | dict):
|
|
user_id, user_email = self._get_user(session)
|
|
return self._db_engine.put_many(user_id, user_email, entry, items)
|
|
|
|
def get(self, session: dict, entry: str, key: str | None = None, default=NoDefault):
|
|
try:
|
|
user_id, _ = self._get_user(session)
|
|
return self._db_engine.get(user_id, entry, key)
|
|
except DbException:
|
|
if default is NoDefault:
|
|
raise
|
|
else:
|
|
return default
|
|
|
|
def exists(self, session: dict, entry: str):
|
|
user_id, _ = self._get_user(session)
|
|
|
|
return self._db_engine.exists(user_id, entry)
|
|
|
|
def get_digest(self, session: dict, entry: str):
|
|
user_id, _ = self._get_user(session)
|
|
return self._db_engine.get_digest(user_id, entry)
|
|
|
|
def history(self, session, entry, digest=None, max_items=1000):
|
|
user_id, _ = self._get_user(session)
|
|
return self._db_engine.history(user_id, entry, digest, max_items)
|
|
|
|
def get_db_engine(self):
|
|
return self._db_engine
|
|
|
|
@staticmethod
|
|
def _get_user(session):
|
|
user_id = str(session.get("user_id", NOT_LOGGED)) if session else NO_SESSION
|
|
user_email = session.get("user_email", NOT_LOGGED) if session else NO_SESSION
|
|
return user_id, user_email
|
|
|
|
|
|
class SettingsTransaction:
|
|
def __init__(self, session, settings_manager: SettingsManager):
|
|
self._settings_manager = settings_manager
|
|
self._session = session
|
|
self._user_id = session["user_id"] if session else NO_SESSION
|
|
self._user_email = session["user_email"] if session else NOT_LOGGED
|
|
self._entries = None
|
|
|
|
def __enter__(self):
|
|
self._entries = self._settings_manager.load(self._user_email, self._user_id)
|
|
return self
|
|
|
|
def put(self, key: str, value: object):
|
|
self._entries[key] = value
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if exc_type is None:
|
|
self._settings_manager.save(self._user_email, self._user_id, self._entries)
|
|
|
|
|
|
class GenericDbManager:
|
|
"""
|
|
Given an obj_entry (entry in DbEngine) and obj_type (object to serialize),
|
|
>>> db = GenericDbManager(session, settings_manager, obj_entry, obj_type)
|
|
>>> db.prop_name = value # will save the value to the database.
|
|
>>> db.prop_name # will load the value from the database.
|
|
|
|
"""
|
|
|
|
def __init__(self, session, settings_manager: SettingsManager, obj_entry, obj_type):
|
|
self.__dict__["_session"] = session
|
|
self.__dict__["_settings_manager"] = settings_manager
|
|
self.__dict__["_obj_entry"] = obj_entry
|
|
self.__dict__["_obj_type"] = obj_type
|
|
|
|
def __setattr__(self, key, value):
|
|
if key.startswith("_"):
|
|
super().__setattr__(key, value)
|
|
|
|
settings = self._settings_manager.load(self._session, self._obj_entry, default=self._obj_type())
|
|
if not (hasattr(settings, key)):
|
|
raise AttributeError(f"Settings '{self._obj_entry}' has no attribute '{key}'.")
|
|
|
|
setattr(settings, key, value)
|
|
self._settings_manager.save(self._session, self._obj_entry, settings)
|
|
|
|
def __getattr__(self, item):
|
|
if item.startswith("_"):
|
|
return super().__getattribute__(item)
|
|
|
|
settings = self._settings_manager.load(self._session, self._obj_entry, default=self._obj_type())
|
|
if not (hasattr(settings, item)):
|
|
raise AttributeError(f"Settings '{self._obj_entry}' has no attribute '{item}'.")
|
|
|
|
return getattr(settings, item)
|
|
|
|
|
|
class NestedSettingsManager:
|
|
"""
|
|
Manages access and modification of a specific subset of persistent settings.
|
|
|
|
The GenericSubDbManager class provides mechanisms to dynamically get and set
|
|
attributes for a specified settings object accessed through a session. It
|
|
handles the retrieval, modification, and persistence of settings using a
|
|
settings manager, ensuring that only valid attributes can be managed.
|
|
|
|
Given an obj_entry (entry in DbEngine) and obj_type (object to serialize),
|
|
>>> db = NestedSettingsManager(session, settings_manager, obj_entry, obj_type, obj_attribute)
|
|
>>> db.prop_name = value # will save obj_type.obj_attribute.prop_name in database
|
|
>>> db.prop_name # will load obj_type.obj_attribute.prop_name from database
|
|
"""
|
|
|
|
def __init__(self, session: dict, settings_manager: SettingsManager, obj_entry, obj_type, obj_attribute):
|
|
self.__dict__["_session"] = session
|
|
self.__dict__["_settings_manager"] = settings_manager
|
|
self.__dict__["_obj_entry"] = obj_entry
|
|
self.__dict__["_obj_type"] = obj_type
|
|
self.__dict__["_obj_attribute"] = obj_attribute
|
|
|
|
def __getattr__(self, item):
|
|
if item.startswith("_"):
|
|
return super().__getattribute__(item)
|
|
|
|
settings, obj = self._get_settings_and_object()
|
|
|
|
if not hasattr(obj, item):
|
|
raise AttributeError(f"Settings '{self._obj_attribute}' has no attribute '{item}'.")
|
|
|
|
return getattr(obj, item)
|
|
|
|
def __setattr__(self, key, value):
|
|
if key.startswith("_"):
|
|
super().__setattr__(key, value)
|
|
|
|
settings, obj = self._get_settings_and_object()
|
|
if not (hasattr(obj, key)):
|
|
raise AttributeError(f"Settings '{self._obj_attribute}', from '{self._obj_entry}' has no attribute '{key}'.")
|
|
|
|
setattr(obj, key, value)
|
|
self._settings_manager.save(self._session, self._obj_entry, settings)
|
|
|
|
def update(self, values: dict, ignore_missing=False):
|
|
settings, obj = self._get_settings_and_object()
|
|
for k, v in values.items():
|
|
if hasattr(obj, k):
|
|
setattr(obj, k, v)
|
|
elif not ignore_missing:
|
|
raise AttributeError(f"Settings '{self._obj_attribute}', from '{self._obj_entry}' has no attribute '{k}'.")
|
|
|
|
self._settings_manager.save(self._session, self._obj_entry, settings)
|
|
|
|
def _get_settings_and_object(self):
|
|
settings = self._settings_manager.load(self._session, self._obj_entry, default=self._obj_type())
|
|
if not hasattr(settings, self._obj_attribute):
|
|
raise AttributeError(f"Settings '{self._obj_entry}' has no attribute '{self._obj_attribute}'.")
|
|
|
|
return settings, getattr(settings, self._obj_attribute)
|