kdb-bot/.cpl/schematic_db_table.py

@@ -0,0 +1,100 @@
+from cpl_cli.abc.generate_schematic_abc import GenerateSchematicABC
+
+
+class DBTable(GenerateSchematicABC):
+    def __init__(self, *args: str):
+        GenerateSchematicABC.__init__(self, *args)
+        self._name = self._name.replace("_db_table", "")
+        self._class_name = self._class_name.split("Db_table")[0]
+
+    def get_code(self) -> str:
+        import textwrap
+
+        code = textwrap.dedent(
+            """\
+                from datetime import datetime
+
+                from cpl_core.database import TableABC
+                
+                
+                class $ClassName(TableABC):
+                    def __init__(
+                        self,
+                        value: str,
+                        created_at: datetime = None,
+                        modified_at: datetime = None,
+                        id=0,
+                    ):
+                        self._id = id
+                        self._value = value
+                
+                        TableABC.__init__(self)
+                        self._created_at = created_at if created_at is not None else self._created_at
+                        self._modified_at = modified_at if modified_at is not None else self._modified_at
+                
+                    @property
+                    def value(self) -> str:
+                        return self._value
+                
+                    @value.setter
+                    def value(self, value: str):
+                        self._value = value
+                
+                    @staticmethod
+                    def get_select_all_string() -> str:
+                        return str(
+                            f\"""
+                                SELECT * FROM `$TableName`;
+                            \"""
+                        )
+                
+                    @staticmethod
+                    def get_select_by_id_string(id: int) -> str:
+                        return str(
+                            f\"""
+                                SELECT * FROM `$TableName`
+                                WHERE `Id` = {id};
+                            \"""
+                        )
+                
+                    @property
+                    def insert_string(self) -> str:
+                        return str(
+                            f\"""
+                                INSERT INTO `$TableName` (
+                                    `Value`
+                                ) VALUES (
+                                    {self._value}
+                                );
+                            \"""
+                        )
+                
+                    @property
+                    def udpate_string(self) -> str:
+                        return str(
+                            f\"""
+                                UPDATE `$TableName`
+                                SET `Value` = {self._value}
+                                WHERE `Id` = {self._id};
+                            \"""
+                        )
+                
+                    @property
+                    def delete_string(self) -> str:
+                        return str(
+                            f\"""
+                                DELETE FROM `$TableName`
+                                WHERE `Id` = {self._id};
+                            \"""
+                        )
+            """
+        )
+        return self.build_code_str(
+            code,
+            ClassName=self._class_name,
+            TableName=self._class_name,
+        )
+
+    @classmethod
+    def register(cls):
+        GenerateSchematicABC.register(cls, "db-table", [])

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

@@ -0,0 +1,25 @@
+from abc import ABC, abstractmethod
+
+from bot_data.model.technician_config import TechnicianConfig
+
+
+class TechnicianConfigRepositoryABC(ABC):
+    @abstractmethod
+    def __init__(self):
+        pass
+
+    @abstractmethod
+    def get_technician_config(self) -> TechnicianConfig:
+        pass
+
+    @abstractmethod
+    def add_technician_config(self, technician_config: TechnicianConfig):
+        pass
+
+    @abstractmethod
+    def update_technician_config(self, technician_config: TechnicianConfig):
+        pass
+
+    @abstractmethod
+    def delete_technician_config(self, technician_config: TechnicianConfig):
+        pass

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

@@ -0,0 +1,132 @@
+from datetime import datetime
+
+from cpl_core.database import TableABC
+from cpl_query.extension import List
+
+
+class TechnicianConfig(TableABC):
+    def __init__(
+        self,
+        help_command_reference_url: str,
+        wait_for_restart: int,
+        wait_for_shutdown: int,
+        cache_max_messages: int,
+        technician_ids: List[int],
+        ping_urls: List[str],
+        created_at: datetime = None,
+        modified_at: datetime = None,
+        id=0,
+    ):
+        self._id = id
+        self._help_command_reference_url = help_command_reference_url
+        self._wait_for_restart = wait_for_restart
+        self._wait_for_shutdown = wait_for_shutdown
+        self._cache_max_messages = cache_max_messages
+        self._technician_ids = technician_ids
+        self._ping_urls = ping_urls
+
+        TableABC.__init__(self)
+        self._created_at = created_at if created_at is not None else self._created_at
+        self._modified_at = modified_at if modified_at is not None else self._modified_at
+
+    @property
+    def help_command_reference_url(self) -> str:
+        return self._help_command_reference_url
+
+    @help_command_reference_url.setter
+    def help_command_reference_url(self, value: str):
+        self._help_command_reference_url = value
+
+    @property
+    def wait_for_restart(self) -> int:
+        return self._wait_for_restart
+
+    @wait_for_restart.setter
+    def wait_for_restart(self, value: int):
+        self._wait_for_restart = value
+
+    @property
+    def wait_for_shutdown(self) -> int:
+        return self._wait_for_shutdown
+
+    @wait_for_shutdown.setter
+    def wait_for_shutdown(self, value: int):
+        self._wait_for_shutdown = value
+
+    @property
+    def cache_max_messages(self) -> int:
+        return self._cache_max_messages
+
+    @cache_max_messages.setter
+    def cache_max_messages(self, value: int):
+        self._cache_max_messages = value
+
+    @property
+    def technician_ids(self) -> List[int]:
+        return self._technician_ids
+
+    @technician_ids.setter
+    def technician_ids(self, value: List[int]):
+        self._technician_ids = value
+
+    @property
+    def ping_urls(self) -> List[str]:
+        return self._ping_urls
+
+    @ping_urls.setter
+    def ping_urls(self, value: List[str]):
+        self._ping_urls = value
+
+    @staticmethod
+    def get_select_all_string() -> str:
+        return str(
+            f"""
+                SELECT * FROM `CFG_Technician`;
+            """
+        )
+
+    @staticmethod
+    def get_select_by_id_string(id: int) -> str:
+        return str(
+            f"""
+                SELECT * FROM `CFG_Technician`
+                WHERE `Id` = {id};
+            """
+        )
+
+    @property
+    def insert_string(self) -> str:
+        return str(
+            f"""
+                INSERT INTO `CFG_Technician` (
+                    `HelpCommandReferenceUrl`, `WaitForRestart`, `WaitForShutdown`, `CacheMaxMessages`
+                ) VALUES (
+                    '{self._help_command_reference_url}',
+                    {self._wait_for_restart},
+                    {self._wait_for_shutdown},
+                    {self._cache_max_messages},
+                );
+            """
+        )
+
+    @property
+    def udpate_string(self) -> str:
+        return str(
+            f"""
+                UPDATE `CFG_Technician`
+                SET `HelpCommandReferenceUrl` = '{self._help_command_reference_url}',
+                `WaitForRestart` = {self._wait_for_restart},
+                `WaitForShutdown` = {self._wait_for_shutdown},
+                `CacheMaxMessages` = {self._cache_max_messages}
+                WHERE `Id` = {self._id};
+            """
+        )
+
+    @property
+    def delete_string(self) -> str:
+        return str(
+            f"""
+                DELETE FROM `CFG_Technician`
+                WHERE `Id` = {self._id};
+            """
+        )

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

@@ -0,0 +1,75 @@
+from datetime import datetime
+
+from cpl_core.database import TableABC
+
+
+class TechnicianIdConfig(TableABC):
+    def __init__(
+        self,
+        technician_id: str,
+        created_at: datetime = None,
+        modified_at: datetime = None,
+        id=0,
+    ):
+        self._id = id
+        self._technician_id = technician_id
+
+        TableABC.__init__(self)
+        self._created_at = created_at if created_at is not None else self._created_at
+        self._modified_at = modified_at if modified_at is not None else self._modified_at
+
+    @property
+    def technician_id(self) -> str:
+        return self._technician_id
+
+    @technician_id.setter
+    def technician_id(self, value: str):
+        self._technician_id = value
+
+    @staticmethod
+    def get_select_all_string() -> str:
+        return str(
+            f"""
+                SELECT * FROM `CFG_TechnicianIds`;
+            """
+        )
+
+    @staticmethod
+    def get_select_by_id_string(id: int) -> str:
+        return str(
+            f"""
+                SELECT * FROM `CFG_TechnicianIds`
+                WHERE `Id` = {id};
+            """
+        )
+
+    @property
+    def insert_string(self) -> str:
+        return str(
+            f"""
+                INSERT INTO `CFG_TechnicianIds` (
+                    `TechnicianId`
+                ) VALUES (
+                    '{self._technician_id}',
+                );
+            """
+        )
+
+    @property
+    def udpate_string(self) -> str:
+        return str(
+            f"""
+                UPDATE `CFG_TechnicianIds`
+                SET `TechnicianId` = '{self._technician_id}'
+                WHERE `Id` = {self._id};
+            """
+        )
+
+    @property
+    def delete_string(self) -> str:
+        return str(
+            f"""
+                DELETE FROM `CFG_TechnicianIds`
+                WHERE `Id` = {self._id};
+            """
+        )

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

@@ -0,0 +1,75 @@
+from datetime import datetime
+
+from cpl_core.database import TableABC
+
+
+class TechnicianPingUrlConfig(TableABC):
+    def __init__(
+        self,
+        ping_url: str,
+        created_at: datetime = None,
+        modified_at: datetime = None,
+        id=0,
+    ):
+        self._id = id
+        self._ping_url = ping_url
+
+        TableABC.__init__(self)
+        self._created_at = created_at if created_at is not None else self._created_at
+        self._modified_at = modified_at if modified_at is not None else self._modified_at
+
+    @property
+    def ping_url(self) -> str:
+        return self._ping_url
+
+    @ping_url.setter
+    def ping_url(self, value: str):
+        self._ping_url = value
+
+    @staticmethod
+    def get_select_all_string() -> str:
+        return str(
+            f"""
+                SELECT * FROM `CFG_TechnicianPingUrls`;
+            """
+        )
+
+    @staticmethod
+    def get_select_by_id_string(id: int) -> str:
+        return str(
+            f"""
+                SELECT * FROM `CFG_TechnicianPingUrls`
+                WHERE `Id` = {id};
+            """
+        )
+
+    @property
+    def insert_string(self) -> str:
+        return str(
+            f"""
+                INSERT INTO `CFG_TechnicianPingUrls` (
+                    `URL`
+                ) VALUES (
+                    '{self._ping_url}',
+                );
+            """
+        )
+
+    @property
+    def udpate_string(self) -> str:
+        return str(
+            f"""
+                UPDATE `CFG_TechnicianPingUrls`
+                SET `URL` = '{self._ping_url}'
+                WHERE `Id` = {self._id};
+            """
+        )
+
+    @property
+    def delete_string(self) -> str:
+        return str(
+            f"""
+                DELETE FROM `CFG_TechnicianPingUrls`
+                WHERE `Id` = {self._id};
+            """
+        )

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

@@ -0,0 +1,67 @@
+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.technician_config_repository_abc import TechnicianConfigRepositoryABC
+from bot_data.model.technician_config import TechnicianConfig
+from bot_data.model.technician_id_config import TechnicianIdConfig
+from bot_data.model.technician_ping_url_config import TechnicianPingUrlConfig
+
+
+class TechnicianConfigRepositoryService(TechnicianConfigRepositoryABC):
+    def __init__(self, logger: DatabaseLogger, db_context: DatabaseContextABC):
+        TechnicianConfigRepositoryABC.__init__(self)
+
+        self._logger = logger
+        self._context = db_context
+
+    def _get_technician_ids(self) -> List[int]:
+        ids = List(int)
+        self._logger.trace(__name__, f"Send SQL command: {TechnicianIdConfig.get_select_all_string()}")
+        results = self._context.select(TechnicianIdConfig.get_select_all_string())
+        for result in results:
+            self._logger.trace(__name__, f"Got TechnicianId with id {result[0]}")
+            ids.append(result[1])
+
+        return ids
+
+    def _get_technician_ping_urls(self) -> List[str]:
+        urls = List(str)
+        self._logger.trace(__name__, f"Send SQL command: {TechnicianPingUrlConfig.get_select_all_string()}")
+        results = self._context.select(TechnicianPingUrlConfig.get_select_all_string())
+        for result in results:
+            self._logger.trace(__name__, f"Got TechnicianPingUrl with id {result[0]}")
+            urls.append(result[1])
+
+        return urls
+
+    def _from_result(self, result: tuple) -> TechnicianConfig:
+        return TechnicianConfig(
+            result[1],
+            result[2],
+            result[3],
+            result[4],
+            self._get_technician_ids(),
+            self._get_technician_ping_urls(),
+            result[5],
+            result[6],
+            id=result[0],
+        )
+
+    def get_technician_config(self) -> TechnicianConfig:
+        self._logger.trace(__name__, f"Send SQL command: {TechnicianConfig.get_select_all_string()}")
+        result = self._context.select(TechnicianConfig.get_select_all_string())[0]
+
+        return self._from_result(result)
+
+    def add_technician_config(self, technician_config: TechnicianConfig):
+        self._logger.trace(__name__, f"Send SQL command: {technician_config.insert_string}")
+        self._context.cursor.execute(technician_config.insert_string)
+
+    def update_technician_config(self, technician_config: TechnicianConfig):
+        self._logger.trace(__name__, f"Send SQL command: {technician_config.udpate_string}")
+        self._context.cursor.execute(technician_config.udpate_string)
+
+    def delete_technician_config(self, technician_config: TechnicianConfig):
+        self._logger.trace(__name__, f"Send SQL command: {technician_config.delete_string}")
+        self._context.cursor.execute(technician_config.delete_string)