From c7967c6904928c80d451ac8e248837e3325c17b1 Mon Sep 17 00:00:00 2001 From: Sven Heidemann Date: Wed, 9 Nov 2022 18:49:11 +0100 Subject: [PATCH] Added stats repo #46 --- .../bot_data/abc/statistic_repository_abc.py | 36 ++++++++ .../service/statistic_repository_service.py | 88 +++++++++++++++++++ 2 files changed, 124 insertions(+) create mode 100644 kdb-bot/src/bot_data/abc/statistic_repository_abc.py create mode 100644 kdb-bot/src/bot_data/service/statistic_repository_service.py diff --git a/kdb-bot/src/bot_data/abc/statistic_repository_abc.py b/kdb-bot/src/bot_data/abc/statistic_repository_abc.py new file mode 100644 index 0000000000..bf16055d9a --- /dev/null +++ b/kdb-bot/src/bot_data/abc/statistic_repository_abc.py @@ -0,0 +1,36 @@ +from abc import ABC, abstractmethod +from typing import Optional + +from cpl_query.extension import List + +from bot_data.model.statistic import Statistic + + +class StatisticRepositoryABC(ABC): + + @abstractmethod + def __init__(self): pass + + @abstractmethod + def get_statistics(self) -> List[Statistic]: pass + + @abstractmethod + def get_statistics_by_server_id(self, server_id: int) -> List[Statistic]: pass + + @abstractmethod + def get_statistic_by_id(self, id: int) -> Statistic: pass + + @abstractmethod + def get_statistic_by_name(self, name: str, server_id: int) -> Statistic: pass + + @abstractmethod + def find_statistic_by_name(self, name: str, server_id: int) -> Optional[Statistic]: pass + + @abstractmethod + def add_statistic(self, statistic: Statistic): pass + + @abstractmethod + def update_statistic(self, statistic: Statistic): pass + + @abstractmethod + def delete_statistic(self, statistic: Statistic): pass diff --git a/kdb-bot/src/bot_data/service/statistic_repository_service.py b/kdb-bot/src/bot_data/service/statistic_repository_service.py new file mode 100644 index 0000000000..222bdf3832 --- /dev/null +++ b/kdb-bot/src/bot_data/service/statistic_repository_service.py @@ -0,0 +1,88 @@ +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.statistic_repository_abc import StatisticRepositoryABC +from bot_data.model.statistic import Statistic + + +class StatisticRepositoryService(StatisticRepositoryABC): + + def __init__(self, logger: DatabaseLogger, db_context: DatabaseContextABC, statistics: ServerRepositoryABC): + self._logger = logger + self._context = db_context + + self._statistics = statistics + + StatisticRepositoryABC.__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 _statistic_from_result(self, sql_result: tuple) -> Statistic: + statistic = Statistic( + self._get_value_from_result(sql_result[1]), + self._get_value_from_result(sql_result[2]), + self._get_value_from_result(sql_result[3]), + self._statistics.get_server_by_id(sql_result[4]), + id=self._get_value_from_result(sql_result[0]) + ) + + return statistic + + def get_statistics(self) -> List[Statistic]: + statistics = List(Statistic) + self._logger.trace(__name__, f'Send SQL command: {Statistic.get_select_all_string()}') + results = self._context.select(Statistic.get_select_all_string()) + for result in results: + statistics.append(self._statistic_from_result(result)) + + return statistics + + def get_statistics_by_server_id(self, server_id: int) -> List[Statistic]: + statistics = List(Statistic) + self._logger.trace(__name__, f'Send SQL command: {Statistic.get_select_by_server_string(server_id)}') + results = self._context.select(Statistic.get_select_by_server_string(server_id)) + for result in results: + statistics.append(self._statistic_from_result(result)) + + return statistics + + def get_statistic_by_id(self, id: int) -> Statistic: + self._logger.trace(__name__, f'Send SQL command: {Statistic.get_select_by_id_string(id)}') + result = self._context.select(Statistic.get_select_by_id_string(id))[0] + return self._statistic_from_result(result) + + def get_statistic_by_name(self, name: str, server_id: int) -> Statistic: + self._logger.trace(__name__, f'Send SQL command: {Statistic.get_select_by_name_string(name, server_id)}') + result = self._context.select(Statistic.get_select_by_name_string(name, server_id))[0] + return self._statistic_from_result(result) + + def find_statistic_by_name(self, name: str, server_id: int) -> Optional[Statistic]: + self._logger.trace(__name__, f'Send SQL command: {Statistic.get_select_by_name_string(name, server_id)}') + result = self._context.select(Statistic.get_select_by_name_string(name, server_id)) + if result is None or len(result) == 0: + return None + + result = result[0] + + return self._statistic_from_result(result) + + def add_statistic(self, statistic: Statistic): + self._logger.trace(__name__, f'Send SQL command: {statistic.insert_string}') + self._context.cursor.execute(statistic.insert_string) + + def update_statistic(self, statistic: Statistic): + self._logger.trace(__name__, f'Send SQL command: {statistic.udpate_string}') + self._context.cursor.execute(statistic.udpate_string) + + def delete_statistic(self, statistic: Statistic): + self._logger.trace(__name__, f'Send SQL command: {statistic.delete_string}') + self._context.cursor.execute(statistic.delete_string)