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

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)