src/gismo/startup.py

@@ -18,6 +18,7 @@ from gismo_data.abc.known_user_repository_abc import KnownUserRepositoryABC
 from gismo_data.abc.migration_abc import MigrationABC
 from gismo_data.abc.server_repository_abc import ServerRepositoryABC
 from gismo_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
+from gismo_data.abc.user_joined_voice_channel_abc import UserJoinedVoiceChannelRepositoryABC
 from gismo_data.abc.user_repository_abc import UserRepositoryABC
 from gismo_data.db_context import DBContext
 from gismo_data.migration.initial_migration import InitialMigration
@@ -30,6 +31,7 @@ from gismo_data.service.migration_service import MigrationService
 from gismo_data.service.server_repository_service import \
     ServerRepositoryService
 from gismo_data.service.user_joined_server_repository_service import UserJoinedServerRepositoryService
+from gismo_data.service.user_joined_voice_channel_service import UserJoinedVoiceChannelRepositoryService
 from gismo_data.service.user_repository_service import UserRepositoryService
 from modules.base.base import Base
 from modules.boot_log.boot_log import BootLog
@@ -79,6 +81,7 @@ class Startup(StartupABC):
         services.add_transient(ClientRepositoryABC, ClientRepositoryService)
         services.add_transient(KnownUserRepositoryABC, KnownUserRepositoryService)
         services.add_transient(UserJoinedServerRepositoryABC, UserJoinedServerRepositoryService)
+        services.add_transient(UserJoinedVoiceChannelRepositoryABC, UserJoinedVoiceChannelRepositoryService)
 
         # modules
         services.add_transient(ModuleABC, Database)

src/gismo_data/abc/user_joined_voice_channel_abc.py

@@ -0,0 +1,34 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from cpl_query.extension import List
+from gismo_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
+
+class UserJoinedVoiceChannelRepositoryABC(ABC):
+
+    @abstractmethod
+    def __init__(self): pass
+    
+    @abstractmethod
+    def get_user_joined_voice_channels(self) -> List[UserJoinedVoiceChannel]: pass
+    
+    @abstractmethod
+    def get_user_joined_voice_channel_by_id(self, id: int) -> UserJoinedVoiceChannel: pass
+    
+    @abstractmethod
+    def get_user_joined_voice_channels_by_user_id(self, user_id: int) -> list[UserJoinedVoiceChannel]: pass
+    
+    @abstractmethod
+    def get_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> UserJoinedVoiceChannel: pass
+    
+    @abstractmethod
+    def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[UserJoinedVoiceChannel]: pass
+    
+    @abstractmethod
+    def add_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): pass
+    
+    @abstractmethod
+    def update_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): pass
+    
+    @abstractmethod
+    def delete_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): pass

src/gismo_data/service/user_joined_voice_channel_service.py

@@ -0,0 +1,107 @@
+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.user_repository_abc import UserRepositoryABC
+from gismo_data.model.user import User
+from gismo_data.model.user_joined_voice_channel import UserJoinedVoiceChannel
+
+from gismo_data.abc.user_joined_voice_channel_abc import UserJoinedVoiceChannelRepositoryABC
+
+
+class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryABC):
+
+    def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC, users: UserRepositoryABC):
+        self._logger = logger
+        self._context = db_context
+
+        self._users = users
+
+        UserJoinedVoiceChannelRepositoryABC.__init__(self)
+
+    def get_user_joined_voice_channels(self) -> List[UserJoinedVoiceChannel]:
+        joins = List(UserJoinedVoiceChannel)
+        self._logger.trace(__name__, f'Send SQL command: {UserJoinedVoiceChannel.get_select_all_string()}')
+        results = self._context.select(UserJoinedVoiceChannel.get_select_all_string())
+        for result in results:
+            self._logger.trace(__name__, f'Get user-joined-voice-channel with id {result[0]}')
+            joins.append(UserJoinedVoiceChannel(
+                self._users.get_user_by_id(result[1]),
+                result[2],
+                result[3],
+                result[4],
+                result[5],
+                id=result[0]
+            ))
+
+        return joins
+    
+    def get_user_joined_voice_channel_by_id(self, id: int) -> UserJoinedVoiceChannel:
+        self._logger.trace(__name__, f'Send SQL command: {UserJoinedVoiceChannel.get_select_by_id_string(id)}')
+        result = self._context.select(UserJoinedVoiceChannel.get_select_by_id_string(id))[0]
+        return UserJoinedVoiceChannel(
+            self._users.get_user_by_id(result[1]),
+            result[2],
+            result[3],
+            result[4],
+            result[5],
+            id=result[0]
+        )
+    
+    def get_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[UserJoinedVoiceChannel]:
+        joins = List(UserJoinedVoiceChannel)
+        self._logger.trace(__name__, f'Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}')
+        results = self._context.select(UserJoinedVoiceChannel.get_select_by_user_id_string(user_id))
+        for result in results:
+            joins.append(UserJoinedVoiceChannel(
+                self._users.get_user_by_id(result[1]),
+                result[2],
+                result[3],
+                result[4],
+                result[5],
+                id=result[0]
+            ))
+    
+        return joins
+    
+    def get_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> UserJoinedVoiceChannel:
+        self._logger.trace(__name__, f'Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}')
+        result = self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))[0]
+        return UserJoinedVoiceChannel(
+            self._users.get_user_by_id(result[1]),
+            result[2],
+            result[3],
+            result[4],
+            result[5],
+            id=result[0]
+        )
+    
+    def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[UserJoinedVoiceChannel]:
+        self._logger.trace(__name__, f'Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}')
+        result = self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))
+        if result is None or len(result) == 0:
+            return None
+
+        result = result[0]
+
+        return UserJoinedVoiceChannel(
+            self._users.get_user_by_id(result[1]),
+            result[2],
+            result[3],
+            result[4],
+            result[5],
+            id=result[0]
+        )
+    
+    def add_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
+        self._logger.trace(__name__, f'Send SQL command: {user_joined_voice_channel.insert_string}')
+        self._context.cursor.execute(user_joined_voice_channel.insert_string)
+    
+    def update_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
+        self._logger.trace(__name__, f'Send SQL command: {user_joined_voice_channel.udpate_string}')
+        self._context.cursor.execute(user_joined_voice_channel.udpate_string)
+    
+    def delete_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
+        self._logger.trace(__name__, f'Send SQL command: {user_joined_voice_channel.delete_string}')
+        self._context.cursor.execute(user_joined_voice_channel.delete_string)