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

@@ -0,0 +1,35 @@
+from abc import ABC, abstractmethod
+
+from cpl_query.extension import List
+
+from bot_data.model.user_warnings import UserWarnings
+
+
+class UserWarningsRepositoryABC(ABC):
+    @abstractmethod
+    def __init__(self):
+        pass
+
+    @abstractmethod
+    def get_user_warnings(self) -> List[UserWarnings]:
+        pass
+
+    @abstractmethod
+    def get_user_warnings_by_id(self, id: int) -> UserWarnings:
+        pass
+
+    @abstractmethod
+    def get_user_warnings_by_user_id(self, user_id: int) -> List[UserWarnings]:
+        pass
+
+    @abstractmethod
+    def add_user_warnings(self, user_warnings: UserWarnings):
+        pass
+
+    @abstractmethod
+    def update_user_warnings(self, user_warnings: UserWarnings):
+        pass
+
+    @abstractmethod
+    def delete_user_warnings(self, user_warnings: UserWarnings):
+        pass

kdb-bot/src/bot_data/data_module.py

@@ -21,6 +21,7 @@ from bot_data.abc.user_message_count_per_hour_repository_abc import (
     UserMessageCountPerHourRepositoryABC,
 )
 from bot_data.abc.user_repository_abc import UserRepositoryABC
+from bot_data.abc.user_warning_repository_abc import UserWarningsRepositoryABC
 from bot_data.service.api_key_repository_service import ApiKeyRepositoryService
 from bot_data.service.auth_user_repository_service import AuthUserRepositoryService
 from bot_data.service.auto_role_repository_service import AutoRoleRepositoryService
@@ -40,6 +41,7 @@ from bot_data.service.user_message_count_per_hour_repository_service import (
     UserMessageCountPerHourRepositoryService,
 )
 from bot_data.service.user_repository_service import UserRepositoryService
+from bot_data.service.user_warnings_repository_service import UserWarningsRepositoryService
 
 
 class DataModule(ModuleABC):
@@ -61,6 +63,7 @@ class DataModule(ModuleABC):
         services.add_transient(UserJoinedGameServerRepositoryABC, UserJoinedGameServerRepositoryService)
         services.add_transient(AutoRoleRepositoryABC, AutoRoleRepositoryService)
         services.add_transient(LevelRepositoryABC, LevelRepositoryService)
+        services.add_transient(UserWarningsRepositoryABC, UserWarningsRepositoryService)
         services.add_transient(
             UserMessageCountPerHourRepositoryABC,
             UserMessageCountPerHourRepositoryService,

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

@@ -6,7 +6,8 @@ from cpl_core.database import TableABC
 from bot_data.model.user import User
 
 
-class UserWarning(TableABC):
+# had to name it UserWarnings instead of UserWarning because UserWarning is a builtin class
+class UserWarnings(TableABC):
     def __init__(
         self,
         description: str,

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

@@ -0,0 +1,74 @@
+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.user_repository_abc import UserRepositoryABC
+from bot_data.abc.user_warning_repository_abc import UserWarningsRepositoryABC
+from bot_data.model.user_warnings import UserWarnings
+
+
+class UserWarningsRepositoryService(UserWarningsRepositoryABC):
+    def __init__(
+        self,
+        logger: DatabaseLogger,
+        db_context: DatabaseContextABC,
+        users: UserRepositoryABC,
+    ):
+        self._logger = logger
+        self._context = db_context
+
+        self._users = users
+        UserWarningsRepositoryABC.__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 _from_result(self, sql_result: tuple) -> UserWarnings:
+        return UserWarnings(
+            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._get_value_from_result(sql_result[4]),
+            self._get_value_from_result(sql_result[0]),
+        )
+
+    def get_user_warnings(self) -> List[UserWarnings]:
+        warnings = List(UserWarnings)
+        self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_all_string()}")
+        results = self._context.select(UserWarnings.get_select_all_string())
+        for result in results:
+            warnings.append(self._from_result(result))
+
+        return warnings
+
+    def get_user_warnings_by_id(self, id: int) -> UserWarnings:
+        self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_by_id_string(id)}")
+        result = self._context.select(UserWarnings.get_select_by_id_string(id))[0]
+        return self._from_result(result)
+
+    def get_user_warnings_by_user_id(self, user_id: int) -> List[UserWarnings]:
+        warnings = List(UserWarnings)
+        self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_by_user_id_string(user_id)}")
+        results = self._context.select(UserWarnings.get_select_by_user_id_string(user_id))
+        for result in results:
+            warnings.append(self._from_result(result))
+
+        return warnings
+
+    def add_user_warnings(self, user_warnings: UserWarnings):
+        self._logger.trace(__name__, f"Send SQL command: {user_warnings.insert_string}")
+        self._context.cursor.execute(user_warnings.insert_string)
+
+    def update_user_warnings(self, user_warnings: UserWarnings):
+        self._logger.trace(__name__, f"Send SQL command: {user_warnings.udpate_string}")
+        self._context.cursor.execute(user_warnings.udpate_string)
+
+    def delete_user_warnings(self, user_warnings: UserWarnings):
+        self._logger.trace(__name__, f"Send SQL command: {user_warnings.delete_string}")
+        self._context.cursor.execute(user_warnings.delete_string)