src/gismo_data/abc/known_user_repository_abc.py

@@ -0,0 +1,30 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from cpl_query.extension import List
+
+from gismo_data.model.known_user import KnownUser
+
+
+class KnownUserRepositoryABC(ABC):
+
+    @abstractmethod
+    def __init__(self): pass
+    
+    @abstractmethod
+    def get_users(self) -> List[KnownUser]: pass
+    
+    @abstractmethod
+    def get_user_by_id(self, id: int) -> KnownUser: pass
+    
+    @abstractmethod
+    def get_user_by_discord_id(self, discord_id: int) -> KnownUser: pass
+    
+    @abstractmethod
+    def find_user_by_discord_id(self, discord_id: int) -> Optional[KnownUser]: pass
+    
+    @abstractmethod
+    def add_user(self, known_user: KnownUser): pass
+    
+    @abstractmethod
+    def delete_user(self, known_user: KnownUser): pass

src/gismo_data/service/known_user_repository_service.py

@@ -0,0 +1,79 @@
+from typing import Optional
+from cpl_core.database.context import DatabaseContextABC
+from cpl_core.logging import LoggerABC
+from cpl_query.extension import List
+
+from gismo_data.abc.server_repository_abc import ServerRepositoryABC
+from gismo_data.abc.user_repository_abc import UserRepositoryABC
+from gismo_data.model.known_user import KnownUser
+
+
+class UserRepositoryService(UserRepositoryABC):
+
+    def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, servers: ServerRepositoryABC):
+        self._logger = logger
+        self._context = db_context
+
+        self._servers = servers
+
+        UserRepositoryABC.__init__(self)
+
+    def get_users(self) -> List[KnownUser]:
+        users = List(KnownUser)
+        self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_all_string()}')
+        results = self._context.select(KnownUser.get_select_all_string())
+        for result in results:
+            self._logger.trace(__name__, f'Get known_user with id {result[0]}')
+            users.append(KnownUser(
+                result[1],
+                result[2],
+                self._servers.get_server_by_id(result[3]),
+                id=result[0]
+            ))
+
+        return users
+
+    def get_user_by_id(self, id: int) -> KnownUser:
+        self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_by_id_string(id)}')
+        result = self._context.select(KnownUser.get_select_by_id_string(id))
+        return KnownUser(
+            result[1],
+            result[2],
+            self._servers.get_server_by_id(result[3]),
+            id=result[0]
+        )
+
+    def get_user_by_discord_id(self, discord_id: int) -> KnownUser:
+        self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_by_discord_id_string(discord_id)}')
+        result = self._context.select(KnownUser.get_select_by_discord_id_string(discord_id))[0]
+        return KnownUser(
+            result[1],
+            result[2],
+            self._servers.get_server_by_id(result[3]),
+            id=result[0]
+        )
+
+    def find_user_by_discord_id(self, discord_id: int) -> Optional[KnownUser]:
+        self._logger.trace(__name__, f'Send SQL command: {KnownUser.get_select_by_discord_id_string(discord_id)}')
+        result = self._context.select(KnownUser.get_select_by_discord_id_string(discord_id))
+        if result is None or len(result) == 0:
+            return None
+        
+        result = result[0]
+        
+        return KnownUser(
+            result[1],
+            result[2],
+            result[3],
+            result[4],
+            self._servers.get_server_by_id(result[3]),
+            id=result[0]
+        )
+    
+    def add_user(self, known_user: KnownUser):
+        self._logger.trace(__name__, f'Send SQL command: {known_user.insert_string}')
+        self._context.cursor.execute(known_user.insert_string)
+    
+    def delete_user(self, known_user: KnownUser):
+        self._logger.trace(__name__, f'Send SQL command: {known_user.delete_string}')
+        self._context.cursor.execute(known_user.delete_string)