add teams datamodel, service and boilerplate database extension

src/ezgg_lan_manager/services/DatabaseService.py

@@ -14,6 +14,7 @@ from src.ezgg_lan_manager.types.ConfigurationTypes import DatabaseConfiguration
 from src.ezgg_lan_manager.types.News import News
 from src.ezgg_lan_manager.types.Participant import Participant
 from src.ezgg_lan_manager.types.Seat import Seat
+from src.ezgg_lan_manager.types.Team import TeamStatus, Team
 from src.ezgg_lan_manager.types.Ticket import Ticket
 from src.ezgg_lan_manager.types.Tournament import Tournament
 from src.ezgg_lan_manager.types.TournamentBase import GameTitle, TournamentFormat, TournamentStatus, ParticipantType
@@ -967,3 +968,25 @@ class DatabaseService:
                     return await self.remove_participant_from_tournament(participant, tournament)
                 except Exception as e:
                     logger.warning(f"Error removing participant from tournament: {e}")
+
+    async def get_teams(self) -> list[Team]:
+        # ToDo: Implement
+        return []
+
+    async def get_team_by_id(self, team_id: int) -> Optional[Team]:
+        return
+
+    async def get_team_for_user_by_id(self, user_id: int) -> Optional[Team]:
+        return
+
+    async def create_team(self, team_name: str, team_abbr: str, join_password: str) -> Team:
+        return  # ToDo: Raise on existing team
+
+    async def update_team(self, team: Team) -> Team:
+        pass
+
+    async def add_member_to_team(self, team: Team, user: User) -> None:
+        pass
+
+    async def remove_user_from_team(self, team: Team, user: User) -> None:
+        pass

src/ezgg_lan_manager/services/TeamService.py

@@ -0,0 +1,94 @@
+from hashlib import sha256
+from typing import Union, Optional
+from string import ascii_letters, digits
+
+from src.ezgg_lan_manager.services.DatabaseService import DatabaseService
+from src.ezgg_lan_manager.types.User import User
+from src.ezgg_lan_manager.types.Team import TeamStatus, Team
+
+
+class NameNotAllowedError(Exception):
+    def __init__(self, disallowed_char: str) -> None:
+        self.disallowed_char = disallowed_char
+
+class AlreadyMemberError(Exception):
+    def __init__(self) -> None:
+        pass
+
+class NotMemberError(Exception):
+    def __init__(self) -> None:
+        pass
+
+class TeamLeadRemovalError(Exception):
+    def __init__(self) -> None:
+        pass
+
+
+class TeamService:
+    ALLOWED_TEAM_NAME_SYMBOLS = ascii_letters + digits + "!#$%&*+,-./:;<=>?[]^_{|}~"
+
+    def __init__(self, db_service: DatabaseService) -> None:
+        self._db_service = db_service
+
+    async def get_all_teams(self) -> list[Team]:
+        return await self._db_service.get_teams()
+
+    async def get_team_by_id(self, team_id: int) -> Optional[Team]:
+        return await self._db_service.get_team_by_id(team_id)
+
+    async def get_team_for_user_by_id(self, user_id: int) -> Optional[Team]:
+        return await self._db_service.get_team_for_user_by_id(user_id)
+
+    async def create_team(self, team_name: str, team_abbr: str, join_password: str) -> Team:
+        disallowed_char = self._check_for_disallowed_char(team_name)
+        if disallowed_char:
+            raise NameNotAllowedError(disallowed_char)
+        disallowed_char = self._check_for_disallowed_char(team_abbr)
+        if disallowed_char:
+            raise NameNotAllowedError(disallowed_char)
+
+        created_team = await self._db_service.create_team(team_name, team_abbr, join_password)
+        return created_team
+
+    async def update_team(self, team: Team) -> Team:
+        """
+        Updates the team EXCLUDING adding and removing members. This is to be done via add_member_to_team and remove_member_from_team
+        :param team: New instance of Team that is to be updated
+        :return: The modified Team instance
+        """
+        disallowed_char = self._check_for_disallowed_char(team.name)
+        if disallowed_char:
+            raise NameNotAllowedError(disallowed_char)
+        disallowed_char = self._check_for_disallowed_char(team.abbreviation)
+        if disallowed_char:
+            raise NameNotAllowedError(disallowed_char)
+        return await self._db_service.update_team(team)
+
+    async def add_member_to_team(self, team: Team, user: User) -> Team:
+        if user in team.members:
+            raise AlreadyMemberError()
+
+        await self._db_service.add_member_to_team(team, user)
+        return await self.get_team_by_id(team.id)
+
+    async def remove_member_from_team(self, team: Team, user: User) -> Team:
+        if user not in team.members:
+            raise NotMemberError()
+
+        if team.members[user] is TeamStatus.LEADER:
+            raise TeamLeadRemovalError()
+
+        await self._db_service.remove_user_from_team(team, user)
+        return await self.get_team_by_id(team.id)
+
+    async def is_join_password_valid(self, team_id: int, join_password: str) -> bool:
+        team = await self.get_team_by_id(team_id)
+        if not team:
+            return False
+        return team.join_password == join_password
+
+    def _check_for_disallowed_char(self, name: str) -> Optional[str]:
+        for c in name:
+            if c not in self.ALLOWED_TEAM_NAME_SYMBOLS:
+                return c
+        return None

src/ezgg_lan_manager/types/Team.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass
+from enum import Enum
+
+from src.ezgg_lan_manager.types.User import User
+
+class TeamStatus(Enum):
+    MEMBER = 0
+    OFFICER = 1
+    LEADER = 2
+
+
+@dataclass(frozen=True)
+class Team:
+    id: int
+    name: str
+    abbreviation: str
+    members: dict[User, TeamStatus]
+    join_password: str