mirror of
https://github.com/Cog-Creators/Red-DiscordBot.git
synced 2025-12-05 17:02:32 -05:00
[Core] Replaced JsonDB with Config (#770)
This commit is contained in:
Will
Twentysix
521
core/config.py
Normal file
521
core/config.py
Normal file
@@ -0,0 +1,521 @@
|
||||
from pathlib import Path
|
||||
|
||||
from core.drivers.red_json import JSON as JSONDriver
|
||||
from core.drivers.red_mongo import Mongo
|
||||
import logging
|
||||
|
||||
from typing import Callable
|
||||
|
||||
log = logging.getLogger("red.config")
|
||||
|
||||
class BaseConfig:
|
||||
def __init__(self, cog_name, unique_identifier, driver_spawn, force_registration=False,
|
||||
hash_uuid=True, collection="GLOBAL", collection_uuid=None,
|
||||
defaults={}):
|
||||
self.cog_name = cog_name
|
||||
if hash_uuid:
|
||||
self.uuid = str(hash(unique_identifier))
|
||||
else:
|
||||
self.uuid = unique_identifier
|
||||
self.driver_spawn = driver_spawn
|
||||
self._driver = None
|
||||
self.collection = collection
|
||||
self.collection_uuid = collection_uuid
|
||||
|
||||
self.force_registration = force_registration
|
||||
|
||||
try:
|
||||
self.driver.maybe_add_ident(self.uuid)
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
self.driver_getmap = {
|
||||
"GLOBAL": self.driver.get_global,
|
||||
"GUILD": self.driver.get_guild,
|
||||
"CHANNEL": self.driver.get_channel,
|
||||
"ROLE": self.driver.get_role,
|
||||
"USER": self.driver.get_user
|
||||
}
|
||||
|
||||
self.driver_setmap = {
|
||||
"GLOBAL": self.driver.set_global,
|
||||
"GUILD": self.driver.set_guild,
|
||||
"CHANNEL": self.driver.set_channel,
|
||||
"ROLE": self.driver.set_role,
|
||||
"USER": self.driver.set_user
|
||||
}
|
||||
|
||||
self.curr_key = None
|
||||
|
||||
self.unsettable_keys = ("cog_name", "cog_identifier", "_id",
|
||||
"guild_id", "channel_id", "role_id",
|
||||
"user_id", "uuid")
|
||||
self.invalid_keys = (
|
||||
"driver_spawn",
|
||||
"_driver", "collection",
|
||||
"collection_uuid", "force_registration"
|
||||
)
|
||||
|
||||
self.defaults = defaults if defaults else {
|
||||
"GLOBAL": {}, "GUILD": {}, "CHANNEL": {}, "ROLE": {},
|
||||
"MEMBER": {}, "USER": {}}
|
||||
|
||||
@classmethod
|
||||
def get_conf(cls, cog_instance: object, unique_identifier: int=0,
|
||||
force_registration: bool=False):
|
||||
"""
|
||||
Gets a config object that cog's can use to safely store data. The
|
||||
backend to this is totally modular and can easily switch between
|
||||
JSON and a DB. However, when changed, all data will likely be lost
|
||||
unless cogs write some converters for their data.
|
||||
|
||||
Positional Arguments:
|
||||
cog_instance - The cog `self` object, can be passed in from your
|
||||
cog's __init__ method.
|
||||
|
||||
Keyword Arguments:
|
||||
unique_identifier - a random integer or string that is used to
|
||||
differentiate your cog from any other named the same. This way we
|
||||
can safely store data for multiple cogs that are named the same.
|
||||
|
||||
YOU SHOULD USE THIS.
|
||||
|
||||
force_registration - A flag which will cause the Config object to
|
||||
throw exceptions if you try to get/set data keys that you have
|
||||
not pre-registered. I highly recommend you ENABLE this as it
|
||||
will help reduce dumb typo errors.
|
||||
"""
|
||||
|
||||
url = None # TODO: get mongo url
|
||||
port = None # TODO: get mongo port
|
||||
|
||||
def spawn_mongo_driver():
|
||||
return Mongo(url, port)
|
||||
|
||||
# TODO: Determine which backend users want, default to JSON
|
||||
|
||||
cog_name = cog_instance.__class__.__name__
|
||||
|
||||
driver_spawn = JSONDriver(cog_name)
|
||||
|
||||
return cls(cog_name=cog_name, unique_identifier=unique_identifier,
|
||||
driver_spawn=driver_spawn, force_registration=force_registration)
|
||||
|
||||
@classmethod
|
||||
def get_core_conf(cls, force_registration: bool=False):
|
||||
core_data_path = Path.cwd() / 'core' / '.data'
|
||||
driver_spawn = JSONDriver("Core", data_path_override=core_data_path)
|
||||
return cls(cog_name="Core", driver_spawn=driver_spawn,
|
||||
unique_identifier=0,
|
||||
force_registration=force_registration)
|
||||
|
||||
@property
|
||||
def driver(self):
|
||||
if self._driver is None:
|
||||
try:
|
||||
self._driver = self.driver_spawn()
|
||||
except TypeError:
|
||||
return self.driver_spawn
|
||||
|
||||
return self._driver
|
||||
|
||||
def __getattr__(self, key):
|
||||
"""This should be used to return config key data as determined by
|
||||
`self.collection` and `self.collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def __setattr__(self, key, value):
|
||||
if 'defaults' in self.__dict__: # Necessary to let the cog load
|
||||
restricted = list(self.defaults[self.collection].keys()) + \
|
||||
list(self.unsettable_keys)
|
||||
if key in restricted:
|
||||
raise ValueError("Not allowed to dynamically set attributes of"
|
||||
" unsettable_keys: {}".format(restricted))
|
||||
else:
|
||||
self.__dict__[key] = value
|
||||
else:
|
||||
self.__dict__[key] = value
|
||||
|
||||
def clear(self):
|
||||
"""Clears all values in the current context ONLY."""
|
||||
raise NotImplemented
|
||||
|
||||
def set(self, key, value):
|
||||
"""This should set config key with value `value` in the
|
||||
corresponding collection as defined by `self.collection` and
|
||||
`self.collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def guild(self, guild):
|
||||
"""This should return a `BaseConfig` instance with the corresponding
|
||||
`collection` and `collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def channel(self, channel):
|
||||
"""This should return a `BaseConfig` instance with the corresponding
|
||||
`collection` and `collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def role(self, role):
|
||||
"""This should return a `BaseConfig` instance with the corresponding
|
||||
`collection` and `collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def member(self, member):
|
||||
"""This should return a `BaseConfig` instance with the corresponding
|
||||
`collection` and `collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def user(self, user):
|
||||
"""This should return a `BaseConfig` instance with the corresponding
|
||||
`collection` and `collection_uuid`."""
|
||||
raise NotImplemented
|
||||
|
||||
def register_global(self, **global_defaults):
|
||||
"""
|
||||
Registers a new dict of global defaults. This function should
|
||||
be called EVERY TIME the cog loads (aka just do it in
|
||||
__init__)!
|
||||
|
||||
:param global_defaults: Each key should be the key you want to
|
||||
access data by and the value is the default value of that
|
||||
key.
|
||||
:return:
|
||||
"""
|
||||
for k, v in global_defaults.items():
|
||||
try:
|
||||
self._register_global(k, v)
|
||||
except KeyError:
|
||||
log.exception("Bad default global key.")
|
||||
|
||||
def _register_global(self, key, default=None):
|
||||
"""Registers a global config key `key`"""
|
||||
if key in self.unsettable_keys:
|
||||
raise KeyError("Attempt to use restricted key: '{}'".format(key))
|
||||
elif not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
self.defaults["GLOBAL"][key] = default
|
||||
|
||||
def register_guild(self, **guild_defaults):
|
||||
"""
|
||||
Registers a new dict of guild defaults. This function should
|
||||
be called EVERY TIME the cog loads (aka just do it in
|
||||
__init__)!
|
||||
|
||||
:param guild_defaults: Each key should be the key you want to
|
||||
access data by and the value is the default value of that
|
||||
key.
|
||||
:return:
|
||||
"""
|
||||
for k, v in guild_defaults.items():
|
||||
try:
|
||||
self._register_guild(k, v)
|
||||
except KeyError:
|
||||
log.exception("Bad default guild key.")
|
||||
|
||||
def _register_guild(self, key, default=None):
|
||||
"""Registers a guild config key `key`"""
|
||||
if key in self.unsettable_keys:
|
||||
raise KeyError("Attempt to use restricted key: '{}'".format(key))
|
||||
elif not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
self.defaults["GUILD"][key] = default
|
||||
|
||||
def register_channel(self, **channel_defaults):
|
||||
"""
|
||||
Registers a new dict of channel defaults. This function should
|
||||
be called EVERY TIME the cog loads (aka just do it in
|
||||
__init__)!
|
||||
|
||||
:param channel_defaults: Each key should be the key you want to
|
||||
access data by and the value is the default value of that
|
||||
key.
|
||||
:return:
|
||||
"""
|
||||
for k, v in channel_defaults.items():
|
||||
try:
|
||||
self._register_channel(k, v)
|
||||
except KeyError:
|
||||
log.exception("Bad default channel key.")
|
||||
|
||||
def _register_channel(self, key, default=None):
|
||||
"""Registers a channel config key `key`"""
|
||||
if key in self.unsettable_keys:
|
||||
raise KeyError("Attempt to use restricted key: '{}'".format(key))
|
||||
elif not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
self.defaults["CHANNEL"][key] = default
|
||||
|
||||
def register_role(self, **role_defaults):
|
||||
"""
|
||||
Registers a new dict of role defaults. This function should
|
||||
be called EVERY TIME the cog loads (aka just do it in
|
||||
__init__)!
|
||||
|
||||
:param role_defaults: Each key should be the key you want to
|
||||
access data by and the value is the default value of that
|
||||
key.
|
||||
:return:
|
||||
"""
|
||||
for k, v in role_defaults.items():
|
||||
try:
|
||||
self._register_role(k, v)
|
||||
except KeyError:
|
||||
log.exception("Bad default role key.")
|
||||
|
||||
def _register_role(self, key, default=None):
|
||||
"""Registers a role config key `key`"""
|
||||
if key in self.unsettable_keys:
|
||||
raise KeyError("Attempt to use restricted key: '{}'".format(key))
|
||||
elif not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
self.defaults["ROLE"][key] = default
|
||||
|
||||
def register_member(self, **member_defaults):
|
||||
"""
|
||||
Registers a new dict of member defaults. This function should
|
||||
be called EVERY TIME the cog loads (aka just do it in
|
||||
__init__)!
|
||||
|
||||
:param member_defaults: Each key should be the key you want to
|
||||
access data by and the value is the default value of that
|
||||
key.
|
||||
:return:
|
||||
"""
|
||||
for k, v in member_defaults.items():
|
||||
try:
|
||||
self._register_member(k, v)
|
||||
except KeyError:
|
||||
log.exception("Bad default member key.")
|
||||
|
||||
def _register_member(self, key, default=None):
|
||||
"""Registers a member config key `key`"""
|
||||
if key in self.unsettable_keys:
|
||||
raise KeyError("Attempt to use restricted key: '{}'".format(key))
|
||||
elif not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
self.defaults["MEMBER"][key] = default
|
||||
|
||||
def register_user(self, **user_defaults):
|
||||
"""
|
||||
Registers a new dict of user defaults. This function should
|
||||
be called EVERY TIME the cog loads (aka just do it in
|
||||
__init__)!
|
||||
|
||||
:param user_defaults: Each key should be the key you want to
|
||||
access data by and the value is the default value of that
|
||||
key.
|
||||
:return:
|
||||
"""
|
||||
for k, v in user_defaults.items():
|
||||
try:
|
||||
self._register_user(k, v)
|
||||
except KeyError:
|
||||
log.exception("Bad default user key.")
|
||||
|
||||
def _register_user(self, key, default=None):
|
||||
"""Registers a user config key `key`"""
|
||||
if key in self.unsettable_keys:
|
||||
raise KeyError("Attempt to use restricted key: '{}'".format(key))
|
||||
elif not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
self.defaults["USER"][key] = default
|
||||
|
||||
|
||||
class Config(BaseConfig):
|
||||
"""
|
||||
Config object created by `Config.get_conf()`
|
||||
|
||||
This configuration object is designed to make backend data
|
||||
storage mechanisms pluggable. It also is designed to
|
||||
help a cog developer make fewer mistakes (such as
|
||||
typos) when dealing with cog data and to make those mistakes
|
||||
apparent much faster in the design process.
|
||||
|
||||
It also has the capability to safely store data between cogs
|
||||
that share the same name.
|
||||
|
||||
There are two main components to this config object. First,
|
||||
you have the ability to get data on a level specific basis.
|
||||
The seven levels available are: global, guild, channel, role,
|
||||
member, user, and misc.
|
||||
|
||||
The second main component is registering default values for
|
||||
data in each of the levels. This functionality is OPTIONAL
|
||||
and must be explicitly enabled when creating the Config object
|
||||
using the kwarg `force_registration=True`.
|
||||
|
||||
Basic Usage:
|
||||
Creating a Config object:
|
||||
Use the `Config.get_conf()` class method to create new
|
||||
Config objects.
|
||||
|
||||
See the `Config.get_conf()` documentation for more
|
||||
information.
|
||||
|
||||
Registering Default Values (optional):
|
||||
You can register default values for data at all levels
|
||||
EXCEPT misc.
|
||||
|
||||
Simply pass in the key/value pairs as keyword arguments to
|
||||
the respective function.
|
||||
|
||||
e.g.: conf_obj.register_global(enabled=True)
|
||||
conf_obj.register_guild(likes_red=True)
|
||||
|
||||
Retrieving data by attributes:
|
||||
Since I registered the "enabled" key in the previous example
|
||||
at the global level I can now do:
|
||||
|
||||
conf_obj.enabled()
|
||||
|
||||
which will retrieve the current value of the "enabled"
|
||||
key, making use of the default of "True". I can also do
|
||||
the same for the guild key "likes_red":
|
||||
|
||||
conf_obj.guild(guild_obj).likes_red()
|
||||
|
||||
If I elected to not register default values, you can provide them
|
||||
when you try to access the key:
|
||||
|
||||
conf_obj.no_default(default=True)
|
||||
|
||||
However if you do not provide a default and you do not register
|
||||
defaults, accessing the attribute will return "None".
|
||||
|
||||
Saving data:
|
||||
This is accomplished by using the `set` function available at
|
||||
every level.
|
||||
|
||||
e.g.: conf_obj.set("enabled", False)
|
||||
conf_obj.guild(guild_obj).set("likes_red", False)
|
||||
|
||||
If `force_registration` was enabled when the config object
|
||||
was created you will only be allowed to save keys that you
|
||||
have registered.
|
||||
|
||||
Misc data is special, use `conf.misc()` and `conf.set_misc(value)`
|
||||
respectively.
|
||||
"""
|
||||
|
||||
def __getattr__(self, key) -> Callable:
|
||||
"""
|
||||
Until I've got a better way to do this I'm just gonna fake __call__
|
||||
|
||||
:param key:
|
||||
:return: lambda function with kwarg
|
||||
"""
|
||||
return self._get_value_from_key(key)
|
||||
|
||||
def _get_value_from_key(self, key) -> Callable:
|
||||
try:
|
||||
default = self.defaults[self.collection][key]
|
||||
except KeyError as e:
|
||||
if self.force_registration:
|
||||
raise AttributeError("Key '{}' not registered!".format(key)) from e
|
||||
default = None
|
||||
|
||||
self.curr_key = key
|
||||
|
||||
if self.collection != "MEMBER":
|
||||
ret = lambda default=default: self.driver_getmap[self.collection](
|
||||
self.cog_name, self.uuid, self.collection_uuid, key,
|
||||
default=default)
|
||||
else:
|
||||
mid, sid = self.collection_uuid
|
||||
ret = lambda default=default: self.driver.get_member(
|
||||
self.cog_name, self.uuid, mid, sid, key,
|
||||
default=default)
|
||||
return ret
|
||||
|
||||
def get(self, key, default=None):
|
||||
"""
|
||||
Included as an alternative to registering defaults.
|
||||
|
||||
:param key:
|
||||
:param default:
|
||||
:return:
|
||||
"""
|
||||
|
||||
try:
|
||||
return getattr(self, key)(default=default)
|
||||
except AttributeError:
|
||||
return
|
||||
|
||||
async def set(self, key, value):
|
||||
# Notice to future developers:
|
||||
# This code was commented to allow users to set keys without having to register them.
|
||||
# That being said, if they try to get keys without registering them
|
||||
# things will blow up. I do highly recommend enforcing the key registration.
|
||||
|
||||
if key in self.unsettable_keys or key in self.invalid_keys:
|
||||
raise KeyError("Restricted key name, please use another.")
|
||||
|
||||
if self.force_registration and key not in self.defaults[self.collection]:
|
||||
raise AttributeError("Key '{}' not registered!".format(key))
|
||||
|
||||
if not key.isidentifier():
|
||||
raise RuntimeError("Invalid key name, must be a valid python variable"
|
||||
" name.")
|
||||
|
||||
if self.collection == "GLOBAL":
|
||||
await self.driver.set_global(self.cog_name, self.uuid, key, value)
|
||||
elif self.collection == "MEMBER":
|
||||
mid, sid = self.collection_uuid
|
||||
await self.driver.set_member(self.cog_name, self.uuid, mid, sid,
|
||||
key, value)
|
||||
elif self.collection in self.driver_setmap:
|
||||
func = self.driver_setmap[self.collection]
|
||||
await func(self.cog_name, self.uuid, self.collection_uuid, key, value)
|
||||
|
||||
async def clear(self):
|
||||
await self.driver_setmap[self.collection](
|
||||
self.cog_name, self.uuid, self.collection_uuid, None, None,
|
||||
clear=True)
|
||||
|
||||
def guild(self, guild):
|
||||
new = type(self)(self.cog_name, self.uuid, self.driver,
|
||||
hash_uuid=False, defaults=self.defaults)
|
||||
new.collection = "GUILD"
|
||||
new.collection_uuid = guild.id
|
||||
new._driver = None
|
||||
return new
|
||||
|
||||
def channel(self, channel):
|
||||
new = type(self)(self.cog_name, self.uuid, self.driver,
|
||||
hash_uuid=False, defaults=self.defaults)
|
||||
new.collection = "CHANNEL"
|
||||
new.collection_uuid = channel.id
|
||||
new._driver = None
|
||||
return new
|
||||
|
||||
def role(self, role):
|
||||
new = type(self)(self.cog_name, self.uuid, self.driver,
|
||||
hash_uuid=False, defaults=self.defaults)
|
||||
new.collection = "ROLE"
|
||||
new.collection_uuid = role.id
|
||||
new._driver = None
|
||||
return new
|
||||
|
||||
def member(self, member):
|
||||
guild = member.guild
|
||||
new = type(self)(self.cog_name, self.uuid, self.driver,
|
||||
hash_uuid=False, defaults=self.defaults)
|
||||
new.collection = "MEMBER"
|
||||
new.collection_uuid = (member.id, guild.id)
|
||||
new._driver = None
|
||||
return new
|
||||
|
||||
def user(self, user):
|
||||
new = type(self)(self.cog_name, self.uuid, self.driver,
|
||||
hash_uuid=False, defaults=self.defaults)
|
||||
new.collection = "USER"
|
||||
new.collection_uuid = user.id
|
||||
new._driver = None
|
||||
return new
|
||||
Reference in New Issue
Block a user