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

Add Tournaments UI (#32)

Browse Source 
Co-authored-by: David Rodenkirchen <drodenkirchen@linetco.com>
Reviewed-on: #32
This commit was merged in pull request #32.
This commit is contained in:
David Rodenkirchen
2026-02-03 23:00:58 +00:00
parent ff5d715a4e
commit 54df84a7da
28 changed files with 1116 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ezgg_lan_manager/services/DatabaseService.py
+180
View File
@@ -1,6 +1,7 @@
import logging
from datetime import date, datetime
from pprint import pprint
from typing import Optional
from decimal import Decimal
@@ -11,8 +12,11 @@ from src.ezgg_lan_manager.types.CateringMenuItem import CateringMenuItem, Cateri
from src.ezgg_lan_manager.types.CateringOrder import CateringMenuItemsWithAmount, CateringOrderStatus
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.Ticket import Ticket
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import GameTitle, TournamentFormat, TournamentStatus, ParticipantType
from src.ezgg_lan_manager.types.Transaction import Transaction
from src.ezgg_lan_manager.types.User import User
@@ -81,6 +85,54 @@ class DatabaseService:
last_updated_at=data[11]
)
@staticmethod
def _parse_tournament_format(format_as_string: str) -> TournamentFormat:
if format_as_string == "SE_BO_1":
return TournamentFormat.SINGLE_ELIMINATION_BO_1
elif format_as_string == "SE_BO_3":
return TournamentFormat.SINGLE_ELIMINATION_BO_3
elif format_as_string == "SE_BO_5":
return TournamentFormat.SINGLE_ELIMINATION_BO_5
elif format_as_string == "DE_BO_1":
return TournamentFormat.DOUBLE_ELIMINATION_BO_1
elif format_as_string == "DE_BO_3":
return TournamentFormat.DOUBLE_ELIMINATION_BO_3
elif format_as_string == "DE_BO_5":
return TournamentFormat.DOUBLE_ELIMINATION_BO_5
elif format_as_string == "FFA":
return TournamentFormat.FFA
else:
# If this happens, database is FUBAR
raise RuntimeError(f"Unknown TournamentFormat: {format_as_string}")
@staticmethod
def _parse_tournament_status(status_as_string: str) -> TournamentStatus:
if status_as_string == "CLOSED":
return TournamentStatus.CLOSED
elif status_as_string == "OPEN":
return TournamentStatus.OPEN
elif status_as_string == "COMPLETED":
return TournamentStatus.COMPLETED
elif status_as_string == "CANCELED":
return TournamentStatus.CANCELED
elif status_as_string == "INVITE_ONLY":
return TournamentStatus.INVITE_ONLY
elif status_as_string == "ONGOING":
return TournamentStatus.ONGOING
else:
# If this happens, database is FUBAR
raise RuntimeError(f"Unknown TournamentStatus: {status_as_string}")
@staticmethod
def _parse_participant_type(participant_type_as_string: str) -> ParticipantType:
if participant_type_as_string == "PLAYER":
return ParticipantType.PLAYER
elif participant_type_as_string == "TEAM":
return ParticipantType.TEAM
else:
# If this happens, database is FUBAR
raise RuntimeError(f"Unknown ParticipantType: {participant_type_as_string}")
async def get_user_by_name(self, user_name: str) -> Optional[User]:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
@@ -787,3 +839,131 @@ class DatabaseService:
return await self.remove_profile_picture(user_id)
except Exception as e:
logger.warning(f"Error deleting user profile picture: {e}")
async def get_all_tournaments(self) -> list[Tournament]:
logger.info(f"Polling Tournaments...")
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.DictCursor) as cursor:
try:
await cursor.execute(
"""
SELECT
/* =======================
Tournament
======================= */
t.id AS tournament_id,
t.name AS tournament_name,
t.description AS tournament_description,
t.format AS tournament_format,
t.start_time,
t.status AS tournament_status,
t.max_participants,
t.created_at,
/* =======================
Game Title
======================= */
gt.id AS game_title_id,
gt.name AS game_title_name,
gt.description AS game_title_description,
gt.web_link AS game_title_web_link,
gt.image_name AS game_title_image_name,
/* =======================
Tournament Participant
======================= */
tp.id AS participant_id,
tp.user_id,
tp.participant_type,
tp.seed,
tp.joined_at
FROM tournaments t
JOIN game_titles gt
ON gt.id = t.game_title_id
LEFT JOIN tournament_participants tp
ON tp.tournament_id = t.id
ORDER BY
t.id,
tp.seed IS NULL,
tp.seed;
"""
)
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_all_tournaments()
except Exception as e:
logger.warning(f"Error getting tournaments: {e}")
tournaments = []
current_tournament: Optional[Tournament] = None
for row in await cursor.fetchall():
if current_tournament is None or current_tournament.id != row["tournament_id"]:
if current_tournament is not None:
tournaments.append(current_tournament)
current_tournament = Tournament(
id_=row["tournament_id"],
name=row["tournament_name"],
description=row["tournament_description"],
game_title=GameTitle(
name=row["game_title_name"],
description=row["game_title_description"],
web_link=row["game_title_web_link"],
image_name=row["game_title_image_name"]
),
format_=self._parse_tournament_format(row["tournament_format"]),
start_time=row["start_time"],
status=self._parse_tournament_status(row["tournament_status"]),
participants=[Participant(id_=row["user_id"], participant_type=self._parse_participant_type(row["participant_type"]))] if row["user_id"] is not None else [],
matches=None, # ToDo: Implement
rounds=[], # ToDo: Implement
max_participants=row["max_participants"]
)
else:
current_tournament.add_participant(
Participant(id_=row["user_id"], participant_type=self._parse_participant_type(row["participant_type"]))
)
else:
tournaments.append(current_tournament)
return tournaments
async def add_participant_to_tournament(self, participant: Participant, tournament: Tournament) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"INSERT INTO tournament_participants (tournament_id, user_id, participant_type) VALUES (%s, %s, %s);",
(tournament.id, participant.id, participant.participant_type.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_participant_to_tournament(participant, tournament)
except Exception as e:
logger.warning(f"Error adding participant to tournament: {e}")
async def remove_participant_from_tournament(self, participant: Participant, tournament: Tournament) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"DELETE FROM tournament_participants WHERE (tournament_id = %s AND user_id = %s);",
(tournament.id, participant.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_participant_from_tournament(participant, tournament)
except Exception as e:
logger.warning(f"Error removing participant from tournament: {e}")