David Rodenkirchen
aec0734937
Co-authored-by: David Rodenkirchen <drodenkirchen@linetco.com> Reviewed-on: #45
135 lines
4.5 KiB
Python
135 lines
4.5 KiB
Python
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
|