Vereins-IT/ezgg-lan-manager
4
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Teams (#45)

Browse Source 
Co-authored-by: David Rodenkirchen <drodenkirchen@linetco.com>
Reviewed-on: #45
This commit is contained in:
David Rodenkirchen 2026-02-15 00:16:55 +00:00 committed by David Rodenkirchen
parent 908bee1e7b
commit deec60347b
16 changed files with 904 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@ -8,13 +8,16 @@ This repository contains the code for the EZGG LAN Manager.
### Prerequisites
- Working Installation of MySQL 5 or latest MariaDB Server (`mariadb-server` for Debian-based Linux, `XAMPP` for Windows)
- Working Installation of MariaDB Server (version `10.6.25` or later)
+ MySQL should work too, but there are no guarantees.
- Python 3.9 or higher
- PyCharm or similar IDE (optional)
### Step 1: Preparing Database
To prepare the database, apply the SQL file located in `sql/create_database.sql` followed by `sql/tournament_patch.sql` to your database server. This is easily accomplished with the MYSQL Workbench, but it can be also done by pipeing the file into the mariadb-server executable.
To prepare the database, apply the SQL file located in `sql/create_database.sql` to your database server. This is easily accomplished with the MYSQL Workbench, but it can be also done by piping the file into the mariadb-server executable.
After creating the database, apply all patches found in `sql/*_patch.sql` in their numeric order.
Optionally, you can now execute the script `create_demo_database_content.py`, found in `src/ezgg_lan_manager/helpers`. Be aware that it can be buggy sometimes, especially if you overwrite existing data.
@ -43,4 +46,4 @@ FLUSH PRIVILEGES;
```
3. Make sure to **NOT** use the default passwords!
4. Apply the `create_database.sql` when starting the MariaDB container for the first time.
5. Apply the `tournament_patch.sql` when starting the MariaDB container for the first time.
5. Apply the patches (`sql/*_patch.sql`) when starting the MariaDB container for the first time.

2
VERSION
View File

@ -1 +1 @@
0.2.2
0.3.0

0
sql/tournament_patch.sql → sql/01-tournament_patch.sql
View File

63
sql/02-teams_patch.sql Normal file
View File

@ -0,0 +1,63 @@
-- =====================================================
-- Teams
-- =====================================================
DROP TABLE IF EXISTS `team_members`;
DROP TABLE IF EXISTS `teams`;
-- -----------------------------------------------------
-- Teams table
-- -----------------------------------------------------
CREATE TABLE `teams` (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
abbreviation VARCHAR(10) NOT NULL,
join_password VARCHAR(255) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uq_team_name (name),
UNIQUE KEY uq_team_abbr (abbreviation)
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci;
-- -----------------------------------------------------
-- Team Members (Junction Table)
-- -----------------------------------------------------
CREATE TABLE `team_members` (
team_id INT NOT NULL,
user_id INT NOT NULL,
status ENUM('MEMBER','OFFICER','LEADER')
NOT NULL DEFAULT 'MEMBER',
joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (team_id, user_id),
CONSTRAINT fk_tm_team
FOREIGN KEY (team_id)
REFERENCES teams(id)
ON DELETE CASCADE,
CONSTRAINT fk_tm_user
FOREIGN KEY (user_id)
REFERENCES users(user_id)
ON DELETE CASCADE
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci;
-- -----------------------------------------------------
-- Indexes
-- -----------------------------------------------------
CREATE INDEX idx_tm_user
ON team_members(user_id);
CREATE INDEX idx_tm_team_status
ON team_members(team_id, status);

5
src/EzggLanManager.py
View File

@ -172,6 +172,11 @@ if __name__ == "__main__":
url_segment="tournament-rules",
build=pages.TournamentRulesPage,
),
ComponentPage(
name="Teams",
url_segment="teams",
build=pages.TeamsPage,
),
ComponentPage(
name="ConwaysGameOfLife",
url_segment="conway",

6
src/ezgg_lan_manager/__init__.py
View File

@ -12,13 +12,14 @@ from src.ezgg_lan_manager.services.MailingService import MailingService
from src.ezgg_lan_manager.services.NewsService import NewsService
from src.ezgg_lan_manager.services.ReceiptPrintingService import ReceiptPrintingService
from src.ezgg_lan_manager.services.SeatingService import SeatingService
from src.ezgg_lan_manager.services.TeamService import TeamService
from src.ezgg_lan_manager.services.TicketingService import TicketingService
from src.ezgg_lan_manager.services.TournamentService import TournamentService
from src.ezgg_lan_manager.services.UserService import UserService
from src.ezgg_lan_manager.types import *
# Inits services in the correct order
def init_services() -> tuple[AccountingService, CateringService, ConfigurationService, DatabaseService, MailingService, NewsService, SeatingService, TicketingService, UserService, LocalDataService, ReceiptPrintingService, TournamentService]:
def init_services() -> tuple[AccountingService, CateringService, ConfigurationService, DatabaseService, MailingService, NewsService, SeatingService, TicketingService, UserService, LocalDataService, ReceiptPrintingService, TournamentService, TeamService]:
logging.basicConfig(level=logging.DEBUG)
configuration_service = ConfigurationService(from_root("config.toml"))
db_service = DatabaseService(configuration_service.get_database_configuration())
@ -32,6 +33,7 @@ def init_services() -> tuple[AccountingService, CateringService, ConfigurationSe
catering_service = CateringService(db_service, accounting_service, user_service, receipt_printing_service)
local_data_service = LocalDataService()
tournament_service = TournamentService(db_service, user_service)
team_service = TeamService(db_service)
return accounting_service, catering_service, configuration_service, db_service, mailing_service, news_service, seating_service, ticketing_service, user_service, local_data_service, receipt_printing_service, tournament_service
return accounting_service, catering_service, configuration_service, db_service, mailing_service, news_service, seating_service, ticketing_service, user_service, local_data_service, receipt_printing_service, tournament_service, team_service

3
src/ezgg_lan_manager/components/DesktopNavigation.py
View File

@ -51,14 +51,13 @@ class DesktopNavigation(Component):
DesktopNavigationButton("Sitzplan", "./seating"),
DesktopNavigationButton("Catering", "./catering"),
DesktopNavigationButton("Teilnehmer", "./guests"),
DesktopNavigationButton("Teams", "./teams"),
DesktopNavigationButton("Turniere", "./tournaments"),
DesktopNavigationButton("FAQ", "./faq"),
DesktopNavigationButton("Regeln & AGB", "./rules-gtc"),
Spacer(min_height=0.7),
DesktopNavigationButton("Discord", "https://discord.gg/8gTjg34yyH", open_new_tab=True),
DesktopNavigationButton("Die EZ GG e.V.", "https://ezgg-ev.de/about", open_new_tab=True),
DesktopNavigationButton("Kontakt", "./contact"),
DesktopNavigationButton("Impressum & DSGVO", "./imprint"),
Spacer(min_height=0.7)
]
team_navigation = [

44
src/ezgg_lan_manager/components/TeamRevealer.py Normal file
View File

@ -0,0 +1,44 @@
from functools import partial
from typing import Callable, Optional, Literal
from rio import Component, Revealer, TextStyle, Column, Row, Tooltip, Icon, Spacer, Text, Button
from src.ezgg_lan_manager.types.Team import TeamStatus, Team
from src.ezgg_lan_manager.types.User import User
class TeamRevealer(Component):
user: Optional[User]
team: Team
mode: Literal["join", "leave", "display"]
on_button_pressed: Callable
def build(self) -> Component:
return Revealer(
header=self.team.name,
header_style=TextStyle(
fill=self.session.theme.background_color,
font_size=1
),
content=Column(
*[Row(
Tooltip(
anchor=Icon("material/star" if self.team.members[member] == TeamStatus.LEADER else "material/stat_1", fill=self.session.theme.hud_color),
tip="Leiter" if self.team.members[member] == TeamStatus.LEADER else "Mitglied", position="top"),
Text(member.user_name, style=TextStyle(fill=self.session.theme.background_color, font_size=1), margin_left=0.5),
Spacer(grow_y=False))
for member in self.team.members
],
Row(Button(
content=f"{self.team.name} beitreten" if self.mode == "join" else f"{self.team.name} verlassen",
shape="rectangle",
style="major",
color="hud",
on_press=partial(self.on_button_pressed, self.team),
), margin_top=1, margin_bottom=1),
margin_right=1,
margin_left=1
),
margin_left=1,
margin_right=1,
)

222
src/ezgg_lan_manager/components/TeamsDialogHandler.py Normal file
View File

@ -0,0 +1,222 @@
import logging
from typing import Optional, Callable
from rio import Component, Text, Spacer, Rectangle, Column, TextStyle, Row, Button, TextInput, ThemeContextSwitcher
from src.ezgg_lan_manager.services.TeamService import TeamService, NotMemberError, TeamLeadRemovalError, AlreadyMemberError, NameNotAllowedError, TeamNameTooLongError, \
TeamAbbrInvalidError, TeamNameAlreadyTaken
from src.ezgg_lan_manager.types.Team import Team
from src.ezgg_lan_manager.types.User import User
logger = logging.getLogger(__name__.split(".")[-1])
class ErrorBox(Component):
error_message: str
cancel: Callable
def build(self) -> Component:
return Rectangle(
content=Column(
Text(self.error_message, style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1.5),
Row(
Button(
content="Ok",
shape="rectangle",
style="major",
color="hud",
on_press=self.cancel,
)
),
margin=1
),
fill=self.session.theme.primary_color
)
class TeamsDialogJoinHandler(Component):
is_active: bool
cancel: Callable
user: Optional[User] = None
team: Optional[Team] = None
error_message: Optional[str] = None
password: str = ""
async def join(self) -> None:
if self.user is None or self.team is None:
return
if self.password != self.team.join_password:
self.error_message = "Falsches Passwort!"
return
try:
await self.session[TeamService].add_member_to_team(self.team, self.user)
except AlreadyMemberError:
self.error_message = "Du bist bereits Mitglied dieses Teams"
else:
await self.cancel_with_reset()
async def cancel_with_reset(self) -> None:
await self.cancel()
self.error_message = None
self.password = ""
def build(self) -> Component:
if not self.is_active or self.user is None or self.team is None:
return Spacer()
if self.error_message is not None:
return ErrorBox(error_message=self.error_message, cancel=self.cancel_with_reset)
return Rectangle(
content=Column(
Text(f"Team {self.team.name} beitreten", style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1, justify="center"),
ThemeContextSwitcher(content=TextInput(text=self.bind().password, label="Beitrittspasswort", margin_bottom=1), color="secondary"),
Row(
Button(
content="Abbrechen",
shape="rectangle",
style="major",
color=self.session.theme.danger_color,
on_press=self.cancel_with_reset,
),
Button(
content="Beitreten",
shape="rectangle",
style="major",
color=self.session.theme.success_color,
on_press=self.join,
),
spacing=1
),
margin=1
),
fill=self.session.theme.primary_color
)
class TeamsDialogLeaveHandler(Component):
is_active: bool
cancel: Callable
user: Optional[User] = None
team: Optional[Team] = None
error_message: Optional[str] = None
async def leave(self) -> None:
if self.user is not None and self.team is not None:
try:
await self.session[TeamService].remove_member_from_team(self.team, self.user)
except NotMemberError:
self.error_message = "Du bist kein Mitglied in diesem Team"
except TeamLeadRemovalError:
self.error_message = "Als Teamleiter kannst du das Team nicht verlassen"
else:
await self.cancel_with_reset()
async def cancel_with_reset(self) -> None:
await self.cancel()
self.error_message = None
def build(self) -> Component:
if not self.is_active or self.user is None or self.team is None:
return Spacer()
if self.error_message is not None:
return ErrorBox(error_message=self.error_message, cancel=self.cancel_with_reset)
return Rectangle(
content=Column(
Text(f"Team {self.team.name} wirklich verlassen?", style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1.5, justify="center"),
Row(
Button(
content="Nein",
shape="rectangle",
style="major",
color=self.session.theme.danger_color,
on_press=self.cancel_with_reset,
),
Button(
content="Ja",
shape="rectangle",
style="major",
color=self.session.theme.success_color,
on_press=self.leave,
),
spacing=1
),
margin=1
),
fill=self.session.theme.primary_color
)
class TeamsDialogCreateHandler(Component):
is_active: bool
cancel: Callable
user: Optional[User] = None
error_message: Optional[str] = None
team_name: str = ""
team_abbr: str = ""
team_join_password: str = ""
async def cancel_with_reset(self) -> None:
await self.cancel()
self.error_message = None
self.team_name, self.team_abbr, self.team_join_password = "", "", ""
async def create(self) -> None:
if self.user is None:
return
if not self.team_name or not self.team_abbr or not self.team_join_password:
self.error_message = "Angaben unvollständig"
return
try:
await self.session[TeamService].create_team(self.team_name, self.team_abbr, self.team_join_password, self.user)
except NameNotAllowedError as e:
self.error_message = f"Angaben ungültig. Darf kein '{e.disallowed_char}' enthalten."
except TeamNameTooLongError:
self.error_message = f"Name zu lang. Maximal {TeamService.MAX_TEAM_NAME_LENGTH} Zeichen."
except TeamAbbrInvalidError:
self.error_message = f"Name zu lang. Maximal {TeamService.MAX_TEAM_ABBR_LENGTH} Zeichen."
except TeamNameAlreadyTaken:
self.error_message = "Ein Team mit diesem Namen existiert bereits."
else:
await self.cancel_with_reset()
def build(self) -> Component:
if not self.is_active or self.user is None:
return Spacer()
if self.error_message is not None:
return ErrorBox(error_message=self.error_message, cancel=self.cancel_with_reset)
return Rectangle(
content=Column(
Text(f"Team gründen", style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1.5, justify="center"),
ThemeContextSwitcher(content=TextInput(text=self.bind().team_name, label="Team Name", margin_bottom=1), color="secondary"),
ThemeContextSwitcher(content=TextInput(text=self.bind().team_abbr, label="Team Abkürzung", margin_bottom=1), color="secondary"),
ThemeContextSwitcher(content=TextInput(text=self.bind().team_join_password, label="Beitrittspasswort", margin_bottom=1), color="secondary"),
Row(
Button(
content="Abbrechen",
shape="rectangle",
style="major",
color=self.session.theme.danger_color,
on_press=self.cancel_with_reset,
),
Button(
content="Gründen",
shape="rectangle",
style="major",
color=self.session.theme.success_color,
on_press=self.create,
),
spacing=1
),
margin=1
),
fill=self.session.theme.primary_color
)

12
src/ezgg_lan_manager/pages/BasePage.py
View File

@ -2,7 +2,7 @@ from __future__ import annotations
from typing import * # type: ignore
from rio import Component, event, Spacer, Card, Container, Column, Row, TextStyle, Color, Text, PageView, Button
from rio import Component, event, Spacer, Card, Container, Column, Row, TextStyle, Color, Text, PageView, Button, Link
from src.ezgg_lan_manager import ConfigurationService, DatabaseService
from src.ezgg_lan_manager.components.DesktopNavigation import DesktopNavigation
@ -58,7 +58,15 @@ class BasePage(Component):
Row(
Spacer(grow_x=True, grow_y=False),
Card(
content=Text(f"EZGG LAN Manager Version {self.session[ConfigurationService].APP_VERSION} © EZ GG e.V.", align_x=0.5, align_y=0.5, fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)),
content=Row(
Text(f"EZGG LAN Manager Version {self.session[ConfigurationService].APP_VERSION} © EZ GG e.V.", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)),
Text(f"-", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5), margin_left=0.5, margin_right=0.5),
Link(content=Text(f"Impressum & DSGVO", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)), target_url="./imprint"),
Text(f"-", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5), margin_left=0.5, margin_right=0.5),
Link(content=Text(f"Kontakt", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)), target_url="./contact"),
align_x=0.5,
align_y=0.5
),
color=self.session.theme.neutral_color,
corner_radius=(0, 0, 0.5, 0.5),
grow_x=False,

172
src/ezgg_lan_manager/pages/TeamsPage.py Normal file
View File

@ -0,0 +1,172 @@
from asyncio import sleep
from rio import event, ProgressCircle, PointerEventListener, PointerEvent, Popup, Color
from src.ezgg_lan_manager import ConfigurationService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.components.TeamRevealer import TeamRevealer
from src.ezgg_lan_manager.components.TeamsDialogHandler import *
from src.ezgg_lan_manager.services.TeamService import TeamService
from src.ezgg_lan_manager.services.UserService import UserService
from src.ezgg_lan_manager.types.SessionStorage import SessionStorage
from src.ezgg_lan_manager.types.Team import Team
from src.ezgg_lan_manager.types.User import User
class TeamsPage(Component):
all_teams: Optional[list[Team]] = None
user: Optional[User] = None
# Dialog handling
popup_open: bool = False
join_active: bool = False
leave_active: bool = True
create_active: bool = False
selected_team_for_join_or_leave: Optional[Team] = None
@event.on_populate
async def on_populate(self) -> None:
self.all_teams = await self.session[TeamService].get_all_teams()
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Teams")
self.session[SessionStorage].subscribe_to_logged_in_or_out_event(str(self.__class__), self.on_populate)
async def on_join_button_pressed(self, team: Team) -> None:
if self.user is None:
return
self.selected_team_for_join_or_leave = team
self.join_active, self.leave_active, self.create_active = True, False, False
self.popup_open = True
async def on_leave_button_pressed(self, team: Team) -> None:
if self.user is None:
return
self.selected_team_for_join_or_leave = team
self.join_active, self.leave_active, self.create_active = False, True, False
self.popup_open = True
async def on_create_button_pressed(self, _: PointerEvent) -> None:
if self.user is None:
return
self.join_active, self.leave_active, self.create_active = False, False, True
self.popup_open = True
async def popup_action_cancelled(self) -> None:
self.popup_open = False
await sleep(0.2) # Waits for the animation to play before resetting its contents
self.join_active, self.leave_active, self.create_active = False, False, False
self.selected_team_for_join_or_leave = None
self.all_teams = await self.session[TeamService].get_all_teams()
def build(self) -> Component:
if self.all_teams is None:
return Column(
MainViewContentBox(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=1,
margin_bottom=1
)
),
Spacer()
)
team_list = []
for team in self.all_teams:
team_list.append(
TeamRevealer(
user=self.user,
team=team,
mode="leave" if self.user in team.members.keys() else "join",
on_button_pressed=self.on_leave_button_pressed if self.user in team.members.keys() else self.on_join_button_pressed
)
)
if team_list:
team_list[-1].margin_bottom = 1
own_teams_content = Spacer(grow_x=False, grow_y=False)
if self.user is not None:
user_team_list = []
for team in self.all_teams:
if self.user in team.members.keys():
user_team_list.append(TeamRevealer(user=self.user, team=team, mode="leave", on_button_pressed=self.on_leave_button_pressed))
if not user_team_list:
user_team_list.append(Text(
text="Du bist noch in keinem Team.",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1
),
margin_top=1,
margin_bottom=1,
align_x=0.5
))
else:
user_team_list[-1].margin_bottom = 1
own_teams_content = MainViewContentBox(
Column(
Row(
Text(
text="Deine Teams",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
grow_x=True,
justify="right",
margin_right=3
),
Column(
PointerEventListener(Rectangle(
content=Text(text="Team erstellen", style=TextStyle(fill=self.session.theme.background_color, font_size=0.7), margin=0.1, selectable=False),
stroke_width=0.1,
stroke_color=self.session.theme.hud_color,
cursor="pointer",
hover_fill=self.session.theme.hud_color,
transition_time=0
), on_press=self.on_create_button_pressed),
Spacer(),
margin_right=2
),
margin_top=1,
margin_bottom=1
),
*user_team_list
)
)
return Popup(
anchor=Column(
own_teams_content,
MainViewContentBox(
Column(
Text(
text="Alle Teams",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=1,
margin_bottom=1,
align_x=0.5
),
*team_list
)
),
align_y=0,
),
content=Column(
TeamsDialogJoinHandler(is_active=self.join_active, cancel=self.popup_action_cancelled, user=self.user, team=self.selected_team_for_join_or_leave),
TeamsDialogLeaveHandler(is_active=self.leave_active, cancel=self.popup_action_cancelled, user=self.user, team=self.selected_team_for_join_or_leave),
TeamsDialogCreateHandler(is_active=self.create_active, cancel=self.popup_action_cancelled, user=self.user)
),
is_open=self.popup_open,
modal=False,
corner_radius=(0.5, 0.5, 0.5, 0.5),
color=Color.TRANSPARENT,
user_closable=False,
position="top"
)

1
src/ezgg_lan_manager/pages/__init__.py
View File

@ -23,3 +23,4 @@ from .OverviewPage import OverviewPage
from .TournamentDetailsPage import TournamentDetailsPage
from .TournamentRulesPage import TournamentRulesPage
from .ConwayPage import ConwayPage
from .TeamsPage import TeamsPage

206
src/ezgg_lan_manager/services/DatabaseService.py
View File

@ -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,208 @@ 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]:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
query = """
SELECT
t.id AS team_id,
t.name AS team_name,
t.abbreviation AS team_abbr,
t.join_password,
t.created_at AS team_created_at,
tm.status AS team_status,
tm.joined_at AS member_joined_at,
u.*
FROM teams t
LEFT JOIN team_members tm
ON t.id = tm.team_id
LEFT JOIN users u
ON tm.user_id = u.user_id
ORDER BY
t.id,
CASE tm.status
WHEN 'LEADER' THEN 1
WHEN 'OFFICER' THEN 2
WHEN 'MEMBER' THEN 3
ELSE 4
END,
u.user_name;
"""
try:
await cursor.execute(query)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.get_teams()
except Exception as e:
logger.warning(f"Error getting teams: {e}")
return []
current_team: Optional[Team] = None
all_teams = []
for row in await cursor.fetchall():
if row[5] is None: # Teams without single member are ignored
continue
if current_team is None:
user = self._map_db_result_to_user(row[7:])
current_team = Team(id=row[0], name=row[1], abbreviation=row[2], join_password=row[3], members={user: TeamStatus.from_str(row[5])})
elif current_team.id == row[0]: # Still same team
current_team.members[self._map_db_result_to_user(row[7:])] = TeamStatus.from_str(row[5])
else:
all_teams.append(current_team)
user = self._map_db_result_to_user(row[7:])
current_team = Team(id=row[0], name=row[1], abbreviation=row[2], join_password=row[3], members={user: TeamStatus.from_str(row[5])})
all_teams.append(current_team)
return all_teams
async def get_team_by_id(self, team_id: int) -> Optional[Team]:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
query = """
SELECT
t.id AS team_id,
t.name AS team_name,
t.abbreviation AS team_abbr,
t.join_password,
t.created_at AS team_created_at,
tm.status AS team_status,
tm.joined_at AS member_joined_at,
u.*
FROM teams t
LEFT JOIN team_members tm
ON t.id = tm.team_id
LEFT JOIN users u
ON tm.user_id = u.user_id
WHERE t.id = %s
ORDER BY
t.id,
CASE tm.status
WHEN 'LEADER' THEN 1
WHEN 'OFFICER' THEN 2
WHEN 'MEMBER' THEN 3
ELSE 4
END,
u.user_name;
"""
try:
await cursor.execute(query, (team_id, ))
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.get_team_by_id(team_id)
except Exception as e:
logger.warning(f"Error getting team: {e}")
return None
team: Optional[Team] = None
for row in await cursor.fetchall():
if team is None:
user = self._map_db_result_to_user(row[7:])
team = Team(id=row[0], name=row[1], abbreviation=row[2], join_password=row[3], members={user: TeamStatus.from_str(row[5])})
elif team.id == row[0]:
team.members[self._map_db_result_to_user(row[7:])] = TeamStatus.from_str(row[5])
return team
async def create_team(self, team_name: str, team_abbr: str, join_password: str, leader: User) -> Team:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"INSERT INTO teams (name, abbreviation, join_password) "
"VALUES (%s, %s, %s)", (team_name, team_abbr, join_password)
)
await conn.commit()
team_id = cursor.lastrowid
await cursor.execute(
"INSERT INTO team_members (team_id, user_id, status) VALUES (%s, %s, %s)",
(team_id, leader.user_id, TeamStatus.LEADER.name)
)
await conn.commit()
return await self.get_team_by_id(team_id)
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.create_team(team_name, team_abbr, join_password)
except aiomysql.IntegrityError as e:
logger.warning(f"Aborted duplication entry: {e}")
raise DuplicationError
async def update_team(self, team: Team) -> Team:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"UPDATE teams SET name = %s, abbreviation = %s, join_password = %s WHERE (id = %s)",
(team.name, team.abbreviation, team.join_password, team.id)
)
await conn.commit()
return await self.get_team_by_id(team.id)
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.update_team(team)
except aiomysql.IntegrityError as e:
logger.warning(f"Aborted duplication entry: {e}")
raise DuplicationError
async def add_member_to_team(self, team: Team, user: User, status: TeamStatus = TeamStatus.MEMBER) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"INSERT INTO team_members (team_id, user_id, status) VALUES (%s, %s, %s)",
(team.id, user.user_id, status.name)
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.add_member_to_team(team, user, status)
except aiomysql.IntegrityError as e:
logger.warning(f"Failed to add member {user.user_name} to team {team.name}: {e}")
raise DuplicationError
async def remove_user_from_team(self, team: Team, user: User) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"DELETE FROM team_members WHERE team_id = %s AND user_id = %s",
(team.id, user.user_id)
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.remove_user_from_team(team, user)

134
src/ezgg_lan_manager/services/TeamService.py Normal file
View File

@ -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

29
src/ezgg_lan_manager/types/Team.py Normal file
View File

@ -0,0 +1,29 @@
from dataclasses import dataclass
from enum import Enum
from typing import Self
from src.ezgg_lan_manager.types.User import User
class TeamStatus(Enum):
MEMBER = 0
OFFICER = 1
LEADER = 2
@classmethod
def from_str(cls, team_status: str) -> Self:
if team_status == "MEMBER":
return TeamStatus.MEMBER
elif team_status == "OFFICER":
return TeamStatus.OFFICER
elif team_status == "LEADER":
return TeamStatus.LEADER
raise ValueError
@dataclass(frozen=True)
class Team:
id: int
name: str
abbreviation: str
members: dict[User, TeamStatus]
join_password: str

7
src/ezgg_lan_manager/types/User.py
View File

@ -19,4 +19,9 @@ class User:
last_updated_at: datetime
def __hash__(self) -> int:
return hash(f"{self.user_id}{self.user_name}{self.user_mail}")
return hash(self.user_id)
def __eq__(self, other):
if not isinstance(other, User):
return NotImplemented
return self.user_id == other.user_id