Added auto role repository #54

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

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
@@ -34,3 +36,4 @@ class DataModule(ModuleABC):
         services.add_transient(KnownUserRepositoryABC, KnownUserRepositoryService)
         services.add_transient(UserJoinedServerRepositoryABC, UserJoinedServerRepositoryService)
         services.add_transient(UserJoinedVoiceChannelRepositoryABC, UserJoinedVoiceChannelRepositoryService)
+        services.add_transient(AutoRoleRepositoryABC, AutoRoleRepositoryService)

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"""

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

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.abc.auto_role_repository_abc import AutoRoleRepositoryABC
-from bot_data.model.server import Server
+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]:
+    def get_auto_roles(self) -> List[AutoRole]:
-        servers = List(Server)
+        auto_roles = List(AutoRole)
-        self._logger.trace(__name__, f'Send SQL command: {Server.get_select_all_string()}')
+        self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_all_string()}')
-        results = self._context.select(Server.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:
+    def get_auto_role_by_id(self, id: int) -> AutoRole:
-        self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_id_string(server_id)}')
+        self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_id_string(id)}')
-        result = self._context.select(Server.get_select_by_id_string(server_id))[0]
+        result = self._context.select(AutoRole.get_select_by_id_string(id))[0]
-        return Server(
+        return AutoRole(
             result[1],
+            result[2],
+            result[3],
+            result[4],
             id=result[0]
         )
 
-    def get_server_by_discord_id(self, discord_id: int) -> Server:
+    def get_auto_roles_by_server_id(self, id: int) -> AutoRole:
-        self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}')
+        self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_server_id_string(id)}')
-        result = self._context.select(Server.get_select_by_discord_id_string(discord_id))[0]
+        result = self._context.select(AutoRole.get_select_by_server_id_string(id))[0]
-        return Server(
+        return AutoRole(
             result[1],
+            result[2],
+            result[3],
+            result[4],
             id=result[0]
         )
 
-    def find_server_by_discord_id(self, discord_id: int) -> Optional[Server]:
+    def find_auto_roles_by_server_id(self, id: int) -> Optional[AutoRole]:
-        self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}')
+        self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_server_id_string(id)}')
-        result = self._context.select(Server.get_select_by_discord_id_string(discord_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 Server(
+        return AutoRole(
             result[1],
             result[2],
             result[3],
+            result[4],
             id=result[0]
         )
 
-    def add_server(self, server: Server):
+    def get_auto_roles_by_message_id(self, id: int) -> AutoRole:
-        self._logger.trace(__name__, f'Send SQL command: {server.insert_string}')
+        self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_message_id_string(id)}')
-        self._context.cursor.execute(server.insert_string)
+        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 update_server(self, server: Server):
+    def find_auto_roles_by_message_id(self, id: int) -> Optional[AutoRole]:
-        self._logger.trace(__name__, f'Send SQL command: {server.udpate_string}')
+        self._logger.trace(__name__, f'Send SQL command: {AutoRole.get_select_by_message_id_string(id)}')
-        self._context.cursor.execute(server.udpate_string)
+        result = self._context.select(AutoRole.get_select_by_message_id_string(id))
+        if result is None or len(result) == 0:
+            return None
 
-    def delete_server(self, server: Server):
+        result = result[0]
-        self._logger.trace(__name__, f'Send SQL command: {server.delete_string}')
+
-        self._context.cursor.execute(server.delete_string)
+        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)