From 8ebfa938277bd174ac1b01765c4819bbcca43994 Mon Sep 17 00:00:00 2001 From: Sven Heidemann Date: Wed, 26 Jul 2023 12:10:24 +0200 Subject: [PATCH] Added technician config repo #127 --- kdb-bot/.cpl/schematic_db_table.py | 100 +++++++++++++ .../abc/technician_config_repository_abc.py | 25 ++++ .../src/bot_data/model/technician_config.py | 132 ++++++++++++++++++ .../bot_data/model/technician_id_config.py | 75 ++++++++++ .../model/technician_ping_url_config.py | 75 ++++++++++ .../technician_config_repository_service.py | 67 +++++++++ 6 files changed, 474 insertions(+) create mode 100644 kdb-bot/.cpl/schematic_db_table.py create mode 100644 kdb-bot/src/bot_data/abc/technician_config_repository_abc.py create mode 100644 kdb-bot/src/bot_data/model/technician_config.py create mode 100644 kdb-bot/src/bot_data/model/technician_id_config.py create mode 100644 kdb-bot/src/bot_data/model/technician_ping_url_config.py create mode 100644 kdb-bot/src/bot_data/service/technician_config_repository_service.py diff --git a/kdb-bot/.cpl/schematic_db_table.py b/kdb-bot/.cpl/schematic_db_table.py new file mode 100644 index 00000000..4c01c549 --- /dev/null +++ b/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", []) diff --git a/kdb-bot/src/bot_data/abc/technician_config_repository_abc.py b/kdb-bot/src/bot_data/abc/technician_config_repository_abc.py new file mode 100644 index 00000000..377103fb --- /dev/null +++ b/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 diff --git a/kdb-bot/src/bot_data/model/technician_config.py b/kdb-bot/src/bot_data/model/technician_config.py new file mode 100644 index 00000000..b90cb493 --- /dev/null +++ b/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}; + """ + ) diff --git a/kdb-bot/src/bot_data/model/technician_id_config.py b/kdb-bot/src/bot_data/model/technician_id_config.py new file mode 100644 index 00000000..ed9c51cc --- /dev/null +++ b/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}; + """ + ) diff --git a/kdb-bot/src/bot_data/model/technician_ping_url_config.py b/kdb-bot/src/bot_data/model/technician_ping_url_config.py new file mode 100644 index 00000000..b9504f5c --- /dev/null +++ b/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}; + """ + ) diff --git a/kdb-bot/src/bot_data/service/technician_config_repository_service.py b/kdb-bot/src/bot_data/service/technician_config_repository_service.py new file mode 100644 index 00000000..de5ea2f2 --- /dev/null +++ b/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)