Add Teams (#45)

Co-authored-by: David Rodenkirchen <drodenkirchen@linetco.com>
Reviewed-on: #45

src/ezgg_lan_manager/services/TeamService.py

@@ -0,0 +1,134 @@
+from string import ascii_letters, digits
+from typing import Optional
+
+from src.ezgg_lan_manager.services.DatabaseService import DatabaseService, DuplicationError
+from src.ezgg_lan_manager.types.Team import TeamStatus, Team
+from src.ezgg_lan_manager.types.User import User
+
+
+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 TeamNameTooLongError(Exception):
+    def __init__(self) -> None:
+        pass
+
+
+class TeamNameAlreadyTaken(Exception):
+    def __init__(self) -> None:
+        pass
+
+
+class TeamAbbrInvalidError(Exception):
+    def __init__(self) -> None:
+        pass
+
+
+class TeamService:
+    ALLOWED_TEAM_NAME_SYMBOLS = ascii_letters + digits + "!#$%&*+,-./:;<=>?[]^_{|}~ "
+    MAX_TEAM_NAME_LENGTH = 24
+    MAX_TEAM_ABBR_LENGTH = 8
+
+    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_teams_for_user_by_id(self, user_id: int) -> list[Team]:
+        all_teams = await self.get_all_teams()
+        user_teams = []
+        for team in all_teams:
+            if user_id in [u.user_id for u in team.members.keys()]:
+                user_teams.append(team)
+        return user_teams
+
+    async def create_team(self, team_name: str, team_abbr: str, join_password: str, leader: User) -> 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)
+
+        if not team_name or len(team_name) > self.MAX_TEAM_NAME_LENGTH:
+            raise TeamNameTooLongError()
+
+        if not team_abbr or len(team_abbr) > self.MAX_TEAM_ABBR_LENGTH:
+            raise TeamAbbrInvalidError()
+
+        try:
+            created_team = await self._db_service.create_team(team_name, team_abbr, join_password, leader)
+        except DuplicationError:
+            raise TeamNameAlreadyTaken
+        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)
+
+        if not team.name or len(team.name) > self.MAX_TEAM_NAME_LENGTH:
+            raise TeamNameTooLongError()
+
+        if not team.abbreviation or len(team.abbreviation) > self.MAX_TEAM_ABBR_LENGTH:
+            raise TeamAbbrInvalidError()
+
+        return await self._db_service.update_team(team)
+
+    async def add_member_to_team(self, team: Team, user: User, status: TeamStatus = TeamStatus.MEMBER) -> Team:
+        if user in team.members:
+            raise AlreadyMemberError()
+
+        await self._db_service.add_member_to_team(team, user, status)
+        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] == 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