profitroll/QuizBot
1
0
Fork 0
Code Issues 5 Pull Requests Actions Projects 1 Releases 2 Activity

Closes #13

Browse Source
This commit is contained in:
Profitroll
2025-05-06 20:16:44 +02:00
parent d1498f38e9
commit 96c1314234
5 changed files with 33 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
classes/base/base_cacheable.py
View File

@@ -2,6 +2,7 @@ from abc import ABC
from logging import Logger
from typing import Optional, Any, Dict
from bson import ObjectId
from libbot.cache.classes import Cache
from classes.abstract import Cacheable
@@ -13,6 +14,8 @@ logger: Logger = get_logger(__name__)
class BaseCacheable(Cacheable, ABC):
"""Base implementation of Cacheable used by all cachable classes."""
_id: ObjectId
async def _set(self, cache: Optional[Cache] = None, **kwargs: Any) -> None:
for key, value in kwargs.items():
if not hasattr(self, key):

91
classes/pycord_event_stage.py
View File

@@ -10,7 +10,7 @@ from discord import File
from libbot.cache.classes import Cache
from pymongo.results import InsertOneResult
from classes.abstract import Cacheable
from classes.base.base_cacheable import BaseCacheable
from classes.errors import EventStageNotFoundError
from modules.database import col_stages
from modules.utils import get_logger, restore_from_cache
@@ -19,7 +19,7 @@ logger: Logger = get_logger(__name__)
@dataclass
class PycordEventStage(Cacheable):
class PycordEventStage(BaseCacheable):
__slots__ = (
"_id",
"event_id",
@@ -110,68 +110,19 @@ class PycordEventStage(Cacheable):
return cls(**db_entry)
async def _set(self, cache: Optional[Cache] = None, **kwargs: Any) -> None:
"""Set attribute data and save it into the database.
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
**kwargs (Any): Mapping of attribute names and respective values to be set
"""
for key, value in kwargs.items():
if not hasattr(self, key):
raise AttributeError()
setattr(self, key, value)
await self.__collection__.update_one({"_id": self._id}, {"$set": kwargs})
self._update_cache(cache)
logger.info("Set attributes of event stage %s to %s", self._id, kwargs)
await super()._set(cache, **kwargs)
async def _remove(self, *args: str, cache: Optional[Cache] = None) -> None:
"""Remove attribute data and save it into the database.
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
*args (str): List of attributes to remove
"""
attributes: Dict[str, Any] = {}
for key in args:
if not hasattr(self, key):
raise AttributeError()
default_value: Any = self.get_default_value(key)
setattr(self, key, default_value)
attributes[key] = default_value
await self.__collection__.update_one({"_id": self._id}, {"$set": attributes})
self._update_cache(cache)
logger.info("Reset attributes %s of event stage %s to default values", args, self._id)
await super()._remove(*args, cache=cache)
def _get_cache_key(self) -> str:
return f"{self.__short_name__}_{self._id}"
def _update_cache(self, cache: Optional[Cache] = None) -> None:
if cache is None:
return
object_dict: Dict[str, Any] = self.to_dict(json_compatible=True)
if object_dict is not None:
cache.set_json(self._get_cache_key(), object_dict)
else:
self._delete_cache(cache)
super()._update_cache(cache)
def _delete_cache(self, cache: Optional[Cache] = None) -> None:
if cache is None:
return
cache.delete(self._get_cache_key())
super()._delete_cache(cache)
@staticmethod
def _entry_to_cache(db_entry: Dict[str, Any]) -> Dict[str, Any]:
@@ -255,41 +206,17 @@ class PycordEventStage(Cacheable):
cache: Optional[Cache] = None,
**kwargs: Any,
) -> None:
"""Update attribute(s) on the object and save the updated entry into the database.
Args:
cache (:obj:`Cache`, optional): Cache engine that will be used to update the cache.
**kwargs (Any): Mapping of attributes in format `attribute_name=attribute_value` to update.
Raises:
AttributeError: Provided attribute does not exist in the class.
"""
await self._set(cache=cache, **kwargs)
await super().update(cache=cache, **kwargs)
async def reset(
self,
*args: str,
cache: Optional[Cache] = None,
) -> None:
"""Remove attribute(s) on the object, replace them with a default value and save the updated entry into the database.
Args:
*args (str): List of attributes to remove.
cache (:obj:`Cache`, optional): Cache engine that will be used to update the cache.
Raises:
AttributeError: Provided attribute does not exist in the class.
"""
await self._remove(*args, cache=cache)
await super().reset(*args, cache=cache)
async def purge(self, cache: Optional[Cache] = None) -> None:
"""Completely remove event stage data from database. Currently only removes the event stage record from events collection.
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
"""
await self.__collection__.delete_one({"_id": self._id})
self._delete_cache(cache)
await super().purge(cache)
# TODO Add documentation
def get_media_files(self) -> List[File] | None:

46
classes/pycord_guild.py
View File

@@ -79,59 +79,18 @@ class PycordGuild(BaseCacheable):
async def _set(self, cache: Optional[Cache] = None, **kwargs: Any) -> None:
await super()._set(cache, **kwargs)
# for key, value in kwargs.items():
# if not hasattr(self, key):
# raise AttributeError()
#
# setattr(self, key, value)
#
# await self.__collection__.update_one({"_id": self._id}, {"$set": kwargs})
#
# self._update_cache(cache)
#
# logger.info("Set attributes of guild %s to %s", self.id, kwargs)
async def _remove(self, *args: str, cache: Optional[Cache] = None) -> None:
await super()._remove(*args, cache=cache)
# attributes: Dict[str, Any] = {}
#
# for key in args:
# if not hasattr(self, key):
# raise AttributeError()
#
# default_value: Any = self.get_default_value(key)
#
# setattr(self, key, default_value)
#
# attributes[key] = default_value
#
# await self.__collection__.update_one({"_id": self._id}, {"$set": attributes})
#
# self._update_cache(cache)
#
# logger.info("Reset attributes %s of guild %s to default values", args, self.id)
def _get_cache_key(self) -> str:
return f"{self.__short_name__}_{self.id}"
def _update_cache(self, cache: Optional[Cache] = None) -> None:
super()._update_cache(cache)
# if cache is None:
# return
#
# object_dict: Dict[str, Any] = self.to_dict(json_compatible=True)
#
# if object_dict is not None:
# cache.set_json(self._get_cache_key(), object_dict)
# else:
# self._delete_cache(cache)
def _delete_cache(self, cache: Optional[Cache] = None) -> None:
super()._delete_cache(cache)
# if cache is None:
# return
#
# cache.delete(self._get_cache_key())
@staticmethod
def _entry_to_cache(db_entry: Dict[str, Any]) -> Dict[str, Any]:
@@ -217,11 +176,6 @@ class PycordGuild(BaseCacheable):
await super().reset(*args, cache=cache)
async def purge(self, cache: Optional[Cache] = None) -> None:
"""Completely remove guild data from database. Currently only removes the guild record from guilds collection.
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
"""
await super().purge(cache)
def is_configured(self) -> bool:

91
classes/pycord_user.py
View File

@@ -17,7 +17,7 @@ from discord.abc import GuildChannel
from libbot.cache.classes import Cache
from pymongo.results import InsertOneResult
from classes.abstract import Cacheable
from classes.base.base_cacheable import BaseCacheable
from classes.errors import (
DiscordCategoryNotFoundError,
DiscordChannelNotFoundError,
@@ -34,17 +34,9 @@ logger: Logger = get_logger(__name__)
@dataclass
class PycordUser(Cacheable):
class PycordUser(BaseCacheable):
"""Dataclass of DB entry of a user"""
# TODO Implement this
async def update(self, cache: Optional[Cache] = None, **kwargs: Any) -> None:
pass
# TODO Implement this
async def reset(self, *args: str, cache: Optional[Cache] = None) -> None:
pass
__slots__ = (
"_id",
"id",
@@ -145,68 +137,19 @@ class PycordUser(Cacheable):
}
async def _set(self, cache: Optional[Cache] = None, **kwargs: Any) -> None:
"""Set attribute data and save it into the database.
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
**kwargs (Any): Mapping of attribute names and respective values to be set
"""
for key, value in kwargs.items():
if not hasattr(self, key):
raise AttributeError()
setattr(self, key, value)
await self.__collection__.update_one({"_id": self._id}, {"$set": kwargs})
self._update_cache(cache)
logger.info("Set attributes of user %s to %s", self.id, kwargs)
await super()._set(cache, **kwargs)
async def _remove(self, *args: str, cache: Optional[Cache] = None) -> None:
"""Remove attribute data and save it into the database.
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
*args (str): List of attributes to remove
"""
attributes: Dict[str, Any] = {}
for key in args:
if not hasattr(self, key):
raise AttributeError()
default_value: Any = self.get_default_value(key)
setattr(self, key, default_value)
attributes[key] = default_value
await self.__collection__.update_one({"_id": self._id}, {"$set": attributes})
self._update_cache(cache)
logger.info("Reset attributes %s of user %s to default values", args, self.id)
await super()._remove(*args, cache=cache)
def _get_cache_key(self) -> str:
return f"{self.__short_name__}_{self.id}_{self.guild_id}"
def _update_cache(self, cache: Optional[Cache] = None) -> None:
if cache is None:
return
object_dict: Dict[str, Any] = self.to_dict(json_compatible=True)
if object_dict is not None:
cache.set_json(self._get_cache_key(), object_dict)
else:
self._delete_cache(cache)
super()._update_cache(cache)
def _delete_cache(self, cache: Optional[Cache] = None) -> None:
if cache is None:
return
cache.delete(self._get_cache_key())
super()._delete_cache(cache)
@staticmethod
def _entry_to_cache(db_entry: Dict[str, Any]) -> Dict[str, Any]:
@@ -270,14 +213,22 @@ class PycordUser(Cacheable):
return PycordUser.get_defaults()[key]
async def purge(self, cache: Optional[Cache] = None) -> None:
"""Completely remove user data from database. Currently only removes the user record from users collection.
async def update(
self,
cache: Optional[Cache] = None,
**kwargs: Any,
) -> None:
await super().update(cache=cache, **kwargs)
Args:
cache (:obj:`Cache`, optional): Cache engine to write the update into
"""
await self.__collection__.delete_one({"_id": self._id})
self._delete_cache(cache)
async def reset(
self,
*args: str,
cache: Optional[Cache] = None,
) -> None:
await super().reset(*args, cache=cache)
async def purge(self, cache: Optional[Cache] = None) -> None:
await super().purge(cache)
# TODO Add documentation
async def event_register(self, event_id: str | ObjectId, cache: Optional[Cache] = None) -> None: