diff --git a/src/bot_data/abc/auto_role_repository_abc.py b/src/bot_data/abc/auto_role_repository_abc.py new file mode 100644 index 00000000..6afa0730 --- /dev/null +++ b/src/bot_data/abc/auto_role_repository_abc.py @@ -0,0 +1,62 @@ +from abc import ABC, abstractmethod +from typing import Optional + +from cpl_query.extension import List + +from bot_data.model.auto_role import AutoRole +from bot_data.model.auto_role_rule import AutoRoleRule +from bot_data.model.server import Server + + +class AutoRoleRepositoryABC(ABC): + + @abstractmethod + def __init__(self): pass + + @abstractmethod + def get_auto_roles(self) -> List[AutoRole]: pass + + @abstractmethod + def get_auto_role_by_id(self, id: int) -> AutoRole: pass + + @abstractmethod + def get_auto_roles_by_server_id(self, id: int) -> AutoRole: pass + + @abstractmethod + def find_auto_roles_by_server_id(self, id: int) -> Optional[AutoRole]: pass + + @abstractmethod + def get_auto_roles_by_message_id(self, id: int) -> AutoRole: pass + + @abstractmethod + def find_auto_roles_by_message_id(self, id: int) -> Optional[AutoRole]: pass + + @abstractmethod + def add_auto_role(self, server: AutoRole): pass + + @abstractmethod + def update_auto_role(self, server: AutoRole): pass + + @abstractmethod + def delete_auto_role(self, server: AutoRole): pass + + @abstractmethod + def get_auto_role_rules(self) -> List[AutoRoleRule]: pass + + @abstractmethod + def get_auto_role_rule_by_id(self, id: int) -> AutoRoleRule: pass + + @abstractmethod + def get_auto_role_rules_by_auto_role_id(self, id: int) -> AutoRoleRule: pass + + @abstractmethod + def find_auto_role_rules_by_auto_role_id(self, id: int) -> Optional[AutoRoleRule]: pass + + @abstractmethod + def add_auto_role_rule(self, server: AutoRoleRule): pass + + @abstractmethod + def update_auto_role_rule(self, server: AutoRoleRule): pass + + @abstractmethod + def delete_auto_role_rule(self, server: AutoRoleRule): pass diff --git a/src/bot_data/data_module.py b/src/bot_data/data_module.py index 099ba2fd..99fc12ad 100644 --- a/src/bot_data/data_module.py +++ b/src/bot_data/data_module.py @@ -5,12 +5,14 @@ from cpl_discord.service.discord_collection_abc import DiscordCollectionABC from bot_core.abc.module_abc import ModuleABC from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum +from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC from bot_data.abc.client_repository_abc import ClientRepositoryABC from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC from bot_data.abc.server_repository_abc import ServerRepositoryABC from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC from bot_data.abc.user_joined_voice_channel_abc import UserJoinedVoiceChannelRepositoryABC from bot_data.abc.user_repository_abc import UserRepositoryABC +from bot_data.service.auto_role_repository_service import AutoRoleRepositoryService from bot_data.service.client_repository_service import ClientRepositoryService from bot_data.service.known_user_repository_service import KnownUserRepositoryService from bot_data.service.server_repository_service import ServerRepositoryService @@ -33,4 +35,5 @@ class DataModule(ModuleABC): services.add_transient(ClientRepositoryABC, ClientRepositoryService) services.add_transient(KnownUserRepositoryABC, KnownUserRepositoryService) services.add_transient(UserJoinedServerRepositoryABC, UserJoinedServerRepositoryService) - services.add_transient(UserJoinedVoiceChannelRepositoryABC, UserJoinedVoiceChannelRepositoryService) \ No newline at end of file + services.add_transient(UserJoinedVoiceChannelRepositoryABC, UserJoinedVoiceChannelRepositoryService) + services.add_transient(AutoRoleRepositoryABC, AutoRoleRepositoryService) diff --git a/src/bot_data/model/auto_role.py b/src/bot_data/model/auto_role.py index 9d02e514..b81c1c76 100644 --- a/src/bot_data/model/auto_role.py +++ b/src/bot_data/model/auto_role.py @@ -35,6 +35,13 @@ class AutoRole(TableABC): WHERE `ServerId` = {id}; """) + @staticmethod + def get_select_by_message_id_string(id: int) -> str: + return str(f""" + SELECT * FROM `AutoRoles` + WHERE `DiscordMessageId` = {id}; + """) + @property def insert_string(self) -> str: return str(f""" diff --git a/src/bot_data/model/auto_role_rule.py b/src/bot_data/model/auto_role_rule.py index f123615a..15ec3e32 100644 --- a/src/bot_data/model/auto_role_rule.py +++ b/src/bot_data/model/auto_role_rule.py @@ -30,10 +30,10 @@ class AutoRoleRule(TableABC): """) @staticmethod - def get_select_by_server_id_string(id: int) -> str: + def get_select_by_auto_role_id_string(id: int) -> str: return str(f""" SELECT * FROM `AutoRoleRules` - WHERE `ServerId` = {id}; + WHERE `AutoRoleId` = {id}; """) @property diff --git a/src/bot_data/service/auto_role_repository_service.py b/src/bot_data/service/auto_role_repository_service.py index 728d2bc6..654d7635 100644 --- a/src/bot_data/service/auto_role_repository_service.py +++ b/src/bot_data/service/auto_role_repository_service.py @@ -4,69 +4,176 @@ from cpl_core.database.context import DatabaseContextABC from cpl_query.extension import List from bot_core.logging.database_logger import DatabaseLogger -from bot_data.abc.server_repository_abc import ServerRepositoryABC -from bot_data.model.server import Server +from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC +from bot_data.model.auto_role import AutoRole +from bot_data.model.auto_role_rule import AutoRoleRule -class ServerRepositoryService(ServerRepositoryABC): +class AutoRoleRepositoryService(AutoRoleRepositoryABC): def __init__(self, logger: DatabaseLogger, db_context: DatabaseContextABC): self._logger = logger self._context = db_context - ServerRepositoryABC.__init__(self) + AutoRoleRepositoryABC.__init__(self) - def get_servers(self) -> List[Server]: - servers = List(Server) - self._logger.trace(__name__, f'Send SQL command: {Server.get_select_all_string()}') - results = self._context.select(Server.get_select_all_string()) + def get_auto_roles(self) -> List[AutoRole]: + auto_roles = List(AutoRole) + self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_all_string()}') + results = self._context.select(AutoRole.get_select_all_string()) for result in results: - servers.append(Server( + auto_roles.append(AutoRole( result[1], + result[2], + result[3], + result[4], id=result[0] )) - return servers + return auto_roles - def get_server_by_id(self, server_id: int) -> Server: - self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_id_string(server_id)}') - result = self._context.select(Server.get_select_by_id_string(server_id))[0] - return Server( - result[1], - id=result[0] - ) - - def get_server_by_discord_id(self, discord_id: int) -> Server: - self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}') - result = self._context.select(Server.get_select_by_discord_id_string(discord_id))[0] - return Server( - result[1], - id=result[0] - ) - - def find_server_by_discord_id(self, discord_id: int) -> Optional[Server]: - self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}') - result = self._context.select(Server.get_select_by_discord_id_string(discord_id)) - if result is None or len(result) == 0: - return None - - result = result[0] - - return Server( + def get_auto_role_by_id(self, id: int) -> AutoRole: + self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_id_string(id)}') + result = self._context.select(AutoRole.get_select_by_id_string(id))[0] + return AutoRole( result[1], result[2], result[3], + result[4], id=result[0] ) - - def add_server(self, server: Server): - self._logger.trace(__name__, f'Send SQL command: {server.insert_string}') - self._context.cursor.execute(server.insert_string) - - def update_server(self, server: Server): - self._logger.trace(__name__, f'Send SQL command: {server.udpate_string}') - self._context.cursor.execute(server.udpate_string) - - def delete_server(self, server: Server): - self._logger.trace(__name__, f'Send SQL command: {server.delete_string}') - self._context.cursor.execute(server.delete_string) + + def get_auto_roles_by_server_id(self, id: int) -> AutoRole: + self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_server_id_string(id)}') + result = self._context.select(AutoRole.get_select_by_server_id_string(id))[0] + return AutoRole( + result[1], + result[2], + result[3], + result[4], + id=result[0] + ) + + def find_auto_roles_by_server_id(self, id: int) -> Optional[AutoRole]: + self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_server_id_string(id)}') + result = self._context.select(AutoRole.get_select_by_server_id_string(id)) + if result is None or len(result) == 0: + return None + + result = result[0] + + return AutoRole( + result[1], + result[2], + result[3], + result[4], + id=result[0] + ) + + def get_auto_roles_by_message_id(self, id: int) -> AutoRole: + self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_message_id_string(id)}') + result = self._context.select(AutoRole.get_select_by_message_id_string(id))[0] + return AutoRole( + result[1], + result[2], + result[3], + result[4], + id=result[0] + ) + + def find_auto_roles_by_message_id(self, id: int) -> Optional[AutoRole]: + self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_message_id_string(id)}') + result = self._context.select(AutoRole.get_select_by_message_id_string(id)) + if result is None or len(result) == 0: + return None + + result = result[0] + + return AutoRole( + result[1], + result[2], + result[3], + result[4], + id=result[0] + ) + + def add_auto_role(self, auto_role: AutoRole): + self._logger.trace(__name__, f'Send SQL command: {auto_role.insert_string}') + self._context.cursor.execute(auto_role.insert_string) + + def update_auto_role(self, auto_role: AutoRole): + self._logger.trace(__name__, f'Send SQL command: {auto_role.udpate_string}') + self._context.cursor.execute(auto_role.udpate_string) + + def delete_auto_role(self, auto_role: AutoRole): + self._logger.trace(__name__, f'Send SQL command: {auto_role.delete_string}') + self._context.cursor.execute(auto_role.delete_string) + + def get_auto_role_rules(self) -> List[AutoRoleRule]: + auto_role_rules = List(AutoRoleRule) + self._logger.trace(__name__, f'Send SQL command: {AutoRoleRule.get_select_all_string()}') + results = self._context.select(AutoRoleRule.get_select_all_string()) + for result in results: + auto_role_rules.append(AutoRoleRule( + result[1], + result[2], + result[3], + result[4], + result[5], + id=result[0] + )) + + return auto_role_rules + + def get_auto_role_rule_by_id(self, id: int) -> AutoRoleRule: + self._logger.trace(__name__, f'Send SQL command: {AutoRoleRule.get_select_by_id_string(id)}') + result = self._context.select(AutoRoleRule.get_select_by_id_string(id))[0] + return AutoRoleRule( + result[1], + result[2], + result[3], + result[4], + result[5], + id=result[0] + ) + + def get_auto_role_rules_by_auto_role_id(self, id: int) -> AutoRoleRule: + self._logger.trace(__name__, f'Send SQL command: {AutoRoleRule.get_select_by_auto_role_id_string(id)}') + result = self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id))[0] + return AutoRoleRule( + result[1], + result[2], + result[3], + result[4], + result[5], + id=result[0] + ) + + def find_auto_role_rules_by_auto_role_id(self, id: int) -> Optional[AutoRoleRule]: + self._logger.trace(__name__, f'Send SQL command: {AutoRoleRule.get_select_by_auto_role_id_string(id)}') + result = self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id)) + if result is None or len(result) == 0: + return None + + result = result[0] + + return AutoRoleRule( + result[1], + result[2], + result[3], + result[4], + result[5], + id=result[0] + ) + + def add_auto_role_rule(self, auto_role_rule: AutoRoleRule): + self._logger.trace(__name__, f'Send SQL command: {auto_role_rule.delete_string}') + self._context.cursor.execute(auto_role_rule.delete_string) + + def update_auto_role_rule(self, auto_role_rule: AutoRoleRule): + self._logger.trace(__name__, f'Send SQL command: {auto_role_rule.delete_string}') + self._context.cursor.execute(auto_role_rule.delete_string) + + def delete_auto_role_rule(self, auto_role_rule: AutoRoleRule): + self._logger.trace(__name__, f'Send SQL command: {auto_role_rule.delete_string}') + self._context.cursor.execute(auto_role_rule.delete_string)