kdb-bot/src/bot_data/abc/short_role_name_repository_abc.py

@@ -0,0 +1,40 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from cpl_query.extension import List
+
+from bot_data.model.short_role_name import ShortRoleName
+
+
+class ShortRoleNameRepositoryABC(ABC):
+    @abstractmethod
+    def __init__(self):
+        pass
+
+    @abstractmethod
+    def get_short_role_names(self) -> List[ShortRoleName]:
+        pass
+
+    @abstractmethod
+    def get_short_role_name_by_id(self, id: int) -> ShortRoleName:
+        pass
+
+    @abstractmethod
+    def find_short_role_name_by_role_id(self, role_id: int) -> Optional[ShortRoleName]:
+        pass
+
+    @abstractmethod
+    def get_short_role_names_by_server_id(self, id: int) -> List[ShortRoleName]:
+        pass
+
+    @abstractmethod
+    def add_short_role_name(self, short_role_name: ShortRoleName):
+        pass
+
+    @abstractmethod
+    def update_short_role_name(self, short_role_name: ShortRoleName):
+        pass
+
+    @abstractmethod
+    def delete_short_role_name(self, short_role_name: ShortRoleName):
+        pass

kdb-bot/src/bot_data/data_module.py

@@ -16,6 +16,7 @@ from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
 from bot_data.abc.level_repository_abc import LevelRepositoryABC
 from bot_data.abc.server_config_repository_abc import ServerConfigRepositoryABC
 from bot_data.abc.server_repository_abc import ServerRepositoryABC
+from bot_data.abc.short_role_name_repository_abc import ShortRoleNameRepositoryABC
 from bot_data.abc.technician_config_repository_abc import TechnicianConfigRepositoryABC
 from bot_data.abc.user_game_ident_repository_abc import UserGameIdentRepositoryABC
 from bot_data.abc.user_joined_game_server_repository_abc import UserJoinedGameServerRepositoryABC
@@ -41,6 +42,7 @@ from bot_data.service.seeder_service import SeederService
 from bot_data.service.server_config_repository_service import ServerConfigRepositoryService
 from bot_data.service.server_config_seeder import ServerConfigSeeder
 from bot_data.service.server_repository_service import ServerRepositoryService
+from bot_data.service.short_role_name_repository_service import ShortRoleNameRepositoryService
 from bot_data.service.technician_config_repository_service import TechnicianConfigRepositoryService
 from bot_data.service.technician_config_seeder import TechnicianConfigSeeder
 from bot_data.service.user_game_ident_repository_service import UserGameIdentRepositoryService
@@ -89,6 +91,7 @@ class DataModule(ModuleABC):
         services.add_transient(AchievementRepositoryABC, AchievementRepositoryService)
         services.add_transient(TechnicianConfigRepositoryABC, TechnicianConfigRepositoryService)
         services.add_transient(ServerConfigRepositoryABC, ServerConfigRepositoryService)
+        services.add_transient(ShortRoleNameRepositoryABC, ShortRoleNameRepositoryService)
 
         services.add_transient(SeederService)
         services.add_transient(DataSeederABC, TechnicianConfigSeeder)

kdb-bot/src/bot_data/model/short_role_name.py

@@ -84,6 +84,15 @@ class ShortRoleName(TableABC):
         """
         )
 
+    @staticmethod
+    def get_select_by_role_id_string(id: int) -> str:
+        return str(
+            f"""
+            SELECT * FROM `ShortRoleNames`
+            WHERE `DiscordRoleId` = {id};
+        """
+        )
+
     @staticmethod
     def get_select_by_server_id_string(id: int) -> str:
         return str(
@@ -98,10 +107,11 @@ class ShortRoleName(TableABC):
         return str(
             f"""
             INSERT INTO `ShortRoleNames` (
-                `Name`, `DiscordRoleId`, `ServerId`
+                `Name`, `DiscordRoleId`, `Position`, `ServerId`
             ) VALUES (
                 '{self._name}',
                 {self._discord_role_id}
+                '{self._position}',
                 {self._server.id},
             );
         """
@@ -114,6 +124,7 @@ class ShortRoleName(TableABC):
             UPDATE `ShortRoleNames`
             SET `Name` = '{self._name}',
             `DiscordRoleId` = {self._discord_role_id},
+            `Position` = '{self._position}',
             `ServerId` = {self._server.id}
             WHERE `Id` = {self._id};
         """

kdb-bot/src/bot_data/service/short_role_name_repository_service.py

@@ -0,0 +1,93 @@
+from typing import Optional
+
+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.short_role_name_repository_abc import ShortRoleNameRepositoryABC
+from bot_data.model.short_role_name import ShortRoleName
+from bot_data.model.short_role_name_position_enum import ShortRoleNamePositionEnum
+
+
+class ShortRoleNameRepositoryService(ShortRoleNameRepositoryABC):
+    def __init__(
+        self,
+        logger: DatabaseLogger,
+        db_context: DatabaseContextABC,
+        servers: ServerRepositoryABC,
+    ):
+        self._logger = logger
+        self._context = db_context
+
+        self._servers = servers
+
+        ShortRoleNameRepositoryABC.__init__(self)
+
+    @staticmethod
+    def _get_value_from_result(value: any) -> Optional[any]:
+        if isinstance(value, str) and "NULL" in value:
+            return None
+
+        return value
+
+    def _short_role_name_from_result(self, sql_result: tuple) -> ShortRoleName:
+        return ShortRoleName(
+            self._get_value_from_result(sql_result[1]),  # name
+            int(self._get_value_from_result(sql_result[2])),  # role_id
+            ShortRoleNamePositionEnum(self._get_value_from_result(sql_result[3])),  # position
+            self._servers.get_server_by_id(sql_result[4]),  # server
+            self._get_value_from_result(sql_result[5]),  # created_at
+            self._get_value_from_result(sql_result[6]),  # modified_at
+            id=self._get_value_from_result(sql_result[0]),  # id
+        )
+
+    def get_short_role_names(self) -> List[ShortRoleName]:
+        short_role_names = List(ShortRoleName)
+        self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_all_string()}")
+        results = self._context.select(ShortRoleName.get_select_all_string())
+        for result in results:
+            self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
+            short_role_names.append(self._short_role_name_from_result(result))
+
+        return short_role_names
+
+    def get_short_role_name_by_id(self, id: int) -> ShortRoleName:
+        self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_by_id_string(id)}")
+        result = self._context.select(ShortRoleName.get_select_by_id_string(id))[0]
+
+        return self._short_role_name_from_result(result)
+
+    def find_short_role_name_by_role_id(self, role_id: int) -> Optional[ShortRoleName]:
+        self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_by_role_id_string(role_id)}")
+        result = self._context.select(ShortRoleName.get_select_by_role_id_string(role_id))
+        if result is None or len(result) == 0:
+            return None
+
+        return self._short_role_name_from_result(result[0])
+
+    def get_short_role_names_by_server_id(self, server_id: int) -> List[ShortRoleName]:
+        short_role_names = List(ShortRoleName)
+        self._logger.trace(
+            __name__,
+            f"Send SQL command: {ShortRoleName.get_select_by_server_id_string(server_id)}",
+        )
+        results = self._context.select(ShortRoleName.get_select_by_server_id_string(server_id))
+
+        for result in results:
+            self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
+            short_role_names.append(self._short_role_name_from_result(result))
+
+        return short_role_names
+
+    def add_short_role_name(self, short_role_name: ShortRoleName):
+        self._logger.trace(__name__, f"Send SQL command: {short_role_name.insert_string}")
+        self._context.cursor.execute(short_role_name.insert_string)
+
+    def update_short_role_name(self, short_role_name: ShortRoleName):
+        self._logger.trace(__name__, f"Send SQL command: {short_role_name.udpate_string}")
+        self._context.cursor.execute(short_role_name.udpate_string)
+
+    def delete_short_role_name(self, short_role_name: ShortRoleName):
+        self._logger.trace(__name__, f"Send SQL command: {short_role_name.delete_string}")
+        self._context.cursor.execute(short_role_name.delete_string)