94 lines
2.8 KiB
Python
94 lines
2.8 KiB
Python
|
import logging
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import List, Union
|
||
|
|
|
||
|
|
from bson import ObjectId, Regex
|
||
|
|
|
||
|
|
from classes.errors import GuildAlreadyExistsError, GuildNotFoundError
|
||
|
|
from classes.pycord_challenge import PycordChallenge
|
||
|
|
from classes.pycord_guild_channels import PycordGuildChannels
|
||
|
|
from classes.pycord_guild_roles import PycordGuildRoles
|
||
|
|
from modules.database import col_guilds
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class PycordGuild:
|
||
|
|
"""Dataclass of DB entry of a member"""
|
||
|
|
|
||
|
|
__slots__ = ("_id", "id", "channels", "roles")
|
||
|
|
|
||
|
|
_id: ObjectId
|
||
|
|
id: int
|
||
|
|
channels: PycordGuildChannels
|
||
|
|
roles: PycordGuildRoles
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def find(cls, id: Union[int, ObjectId]):
|
||
|
|
"""Find guild in the database.
|
||
|
|
|
||
|
|
### Args:
|
||
|
|
* id (`Union[int, ObjectId]`): Guild's Discord ID
|
||
|
|
|
||
|
|
### Raises:
|
||
|
|
* `GuildNotFoundError`: Raised when guild entry could not be found.
|
||
|
|
|
||
|
|
### Returns:
|
||
|
|
* `PycordGuild`: Guild with its database data.
|
||
|
|
"""
|
||
|
|
db_entry = await col_guilds.find_one(
|
||
|
|
({"id": id} if isinstance(id, int) else {"_id": id})
|
||
|
|
)
|
||
|
|
|
||
|
|
if not db_entry:
|
||
|
|
raise GuildNotFoundError(id)
|
||
|
|
|
||
|
|
db_entry["channels"] = PycordGuildChannels(**db_entry["channels"])
|
||
|
|
db_entry["roles"] = PycordGuildRoles(**db_entry["roles"])
|
||
|
|
|
||
|
|
return cls(**db_entry)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def create(
|
||
|
|
cls,
|
||
|
|
id: int,
|
||
|
|
channels: PycordGuildChannels = PycordGuildChannels(),
|
||
|
|
roles: PycordGuildRoles = PycordGuildRoles(),
|
||
|
|
):
|
||
|
|
if await col_guilds.find_one({"id": id}):
|
||
|
|
raise GuildAlreadyExistsError(id)
|
||
|
|
|
||
|
|
guild = {"id": id, "channels": channels.dict(), "roles": roles.dict()}
|
||
|
|
|
||
|
|
inserted = await col_guilds.insert_one(guild)
|
||
|
|
|
||
|
|
guild["_id"] = inserted.inserted_id
|
||
|
|
|
||
|
|
return cls(**guild)
|
||
|
|
|
||
|
|
def is_valid(self) -> bool:
|
||
|
|
"""Check if all attributes are valid and return boolean of that.
|
||
|
|
|
||
|
|
### Returns:
|
||
|
|
* `bool`: `True` if all attributes are valid and `False` if not
|
||
|
|
"""
|
||
|
|
return bool(self.id and self.channels.is_valid() and self.roles.is_valid())
|
||
|
|
|
||
|
|
async def add_challenge(
|
||
|
|
self, challenge: str, answers: List[Regex], enabled: bool
|
||
|
|
) -> PycordChallenge:
|
||
|
|
"""Create new guild's challenge entry in a database.
|
||
|
|
|
||
|
|
### Args:
|
||
|
|
* challenge (`str`): Challenge text
|
||
|
|
* answers (`List[Regex]`): List of regex patterns that are answers to the challenge
|
||
|
|
* enabled (`bool`, *optional*): Whether the challenge is enabled. Defaults to `True`.
|
||
|
|
|
||
|
|
### Returns:
|
||
|
|
* `PycordChallenge`: The challenge object
|
||
|
|
"""
|
||
|
|
return await PycordChallenge.create(
|
||
|
|
self._id, challenge, answers, enabled=enabled
|
||
|
|
)
|