Minidodo
bf6c1842d2
Restart the bot, on heavy DB errors, with an 30 seconds delay (like: all connections terminated by DB, table_definition_cache exhausted) Fix for channel prefixing (org <-> priv)
117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
import inspect
|
|
|
|
from core.decorators import instance
|
|
from core.functions import get_attrs
|
|
from core.logger import Logger
|
|
from core.registry import Registry
|
|
|
|
|
|
@instance()
|
|
class SettingService:
|
|
def __init__(self):
|
|
self.logger = Logger(__name__)
|
|
self.settings = {}
|
|
self.db_cache = {}
|
|
self.change_listeners = {}
|
|
|
|
def inject(self, registry):
|
|
self.db = registry.get_instance("db")
|
|
self.bot = registry.get_instance("bot")
|
|
self.util = registry.get_instance("util")
|
|
|
|
def start(self):
|
|
# process decorators
|
|
for _, inst in Registry.get_all_instances().items():
|
|
for name, method in get_attrs(inst).items():
|
|
if hasattr(method, "setting"):
|
|
key = Registry.get_module_name(inst).split(".")
|
|
# We dont want to load settings, if their modules not enabled in our config...
|
|
if key[0] not in self.bot.modules:
|
|
continue
|
|
setting_name, value, description, extended_description, obj = getattr(method, "setting")
|
|
self.register(inst.module_name, setting_name, value, obj, description, extended_description)
|
|
|
|
def register(self, module, name, value, setting, description, extended_description=None):
|
|
"""Call during start"""
|
|
name = name.lower()
|
|
module = module.lower()
|
|
# do not generate settings for not loaded modules
|
|
if module.split(".")[0] not in self.bot.modules:
|
|
return
|
|
setting.set_name(name)
|
|
setting.set_description(description)
|
|
setting.set_extended_description(extended_description)
|
|
|
|
if not description:
|
|
self.logger.warning(f"No description specified for setting '{name}'")
|
|
|
|
if " " in name:
|
|
raise Exception(f"One or more spaces found in setting name '{name}' for module '{module}'")
|
|
|
|
row = self.db.query_single("SELECT name, value, description FROM setting WHERE name = ?", [name])
|
|
|
|
if row is None:
|
|
self.logger.debug(f"Adding setting '{name}'")
|
|
self.db.exec("INSERT INTO setting (name, value, description, module, verified) VALUES (?, ?, ?, ?, ?)",
|
|
[name, "", description, module, 1])
|
|
|
|
# verify default value is a valid value, and is formatted appropriately
|
|
setting.set_value(value)
|
|
else:
|
|
self.logger.debug(f"Updating setting '{name}'")
|
|
self.db.exec("UPDATE setting SET description = ?, verified = ?, module = ? WHERE name = ?",
|
|
[description, 1, module, name])
|
|
self.settings[name] = setting
|
|
|
|
def register_change_listener(self, setting_name, handler):
|
|
"""
|
|
Call during start
|
|
|
|
Args:
|
|
setting_name: str
|
|
handler: (name: string, old_value, new_value) -> void
|
|
"""
|
|
|
|
if len(inspect.signature(handler).parameters) != 3:
|
|
raise Exception(f"Incorrect number of arguments for handler '{handler.__module__}.{handler.__name__}()'")
|
|
|
|
if setting_name in self.settings:
|
|
if setting_name not in self.change_listeners:
|
|
self.change_listeners[setting_name] = []
|
|
self.change_listeners[setting_name].append(handler)
|
|
else:
|
|
raise Exception(f"Could not register change_listener for setting '{setting_name}' since it does not exist")
|
|
|
|
def get_value(self, name):
|
|
# check cache first
|
|
result = self.db_cache.get(name, None)
|
|
if result:
|
|
return result.value
|
|
else:
|
|
row = self.db.query_single("SELECT value FROM setting WHERE name = ?", [name])
|
|
|
|
# store result in cache
|
|
self.db_cache[name] = row
|
|
|
|
return row.value if row else None
|
|
|
|
def set_value(self, name, value):
|
|
old_value = self.get_value(name)
|
|
|
|
# clear cache
|
|
self.db_cache[name] = None
|
|
|
|
self.db.exec("UPDATE setting SET value = ? WHERE name = ?", [value, name])
|
|
|
|
if name in self.change_listeners:
|
|
for change_listener in self.change_listeners[name]:
|
|
change_listener(name, old_value, value)
|
|
|
|
def get(self, name):
|
|
name = name.lower()
|
|
setting = self.settings.get(name, None)
|
|
if setting:
|
|
return setting
|
|
else:
|
|
return None
|