Added stats repo #46

This commit is contained in:
Sven Heidemann 2022-11-09 18:49:11 +01:00
parent a43676e9bf
commit c7967c6904
2 changed files with 124 additions and 0 deletions

View File

@ -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

View File

@ -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)