Moved a lot of the code into database queries

1 files changed, 164 insertions, 131 deletions

diff --git a/app/model.py b/app/model.py
index b2b6470..2833bb1 100644
--- a/app/model.py
+++ b/app/model.py
@@ -1,83 +1,87 @@
 from math import floor
-from typing import Tuple
-from os.path import isfile
+from app import app, db
 from pandas import read_csv
 from dataclasses import dataclass
-from random import getrandbits, randint
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy import create_engine, not_
+from sqlalchemy.orm import Mapped
+from datetime import date, datetime
+from random import randint, getrandbits
+from typing import Tuple, Optional, Sequence
+from sqlalchemy.orm import mapped_column as mc
 from names import get_first_name, get_last_name
-from app.data import (
-    MIN_LEVEL,
-    MAX_LEVEL,
-    PLAYER_PER_COURT,
-    BASE,
-    GUEST_LEVELS,
-    GUESTS_PER_LEVEL,
-    Player,
-    DisplayPlayer,
-)
+from sqlalchemy import String, Boolean, Integer, ForeignKey, select
 
 
-@dataclass
-class CourtList:
-    total: int = 12
+class User(db.Model):
+    id: Mapped[int] = mc(primary_key=True)
+    username: Mapped[str] = mc(String(64), index=True, unique=True)
+    password_hash: Mapped[Optional[str]] = mc(String(256))
+
+    def __repr__(self):
+        return f"User {self.username}"
+
+
+class Court(db.Model):
+    id: Mapped[int] = mc(primary_key=True)
+    status: Mapped[bool] = mc(Boolean, default=True)
+
+    def __repr__(self):
+        return f"Court {self.id} is {"active" if self.status else "inactive"}"
+
 
-    def __post_init__(self) -> None:
-        self.courts = {i + 1: True for i in range(self.total)}
+class Courts:
 
-    def get_court_status(self, court_number: int) -> bool:
-        return self.courts[court_number]
+    def __init__(self):
+        if app.config["COURT_COUNT"] is not None:
+            db.session.query(Court).delete()
+            for i in range(app.config["COURT_COUNT"]):
+                db.session.add(Court())
+            db.session.commit()
 
-    def toggle_court_status(self, court_number: int) -> bool:
-        new_status = not self.courts[court_number]
-        self.courts[court_number] = new_status
+    def get(self, id: int) -> bool:
+        court = db.session.get(Court, id)
+        if court is None:
+            raise ValueError(f"Court {id} doesn't exist.")
+        return court.status
+
+    def toggle(self, id: int) -> bool:
+        court = db.session.get(Court, id)
+        if court is None:
+            raise ValueError(f"Court {id} doesn't exist.")
+        new_status = not court.status
+        court.status = new_status
+        db.session.commit()
         return new_status
 
-    def get_courts(self) -> dict:
-        return self.courts
+    def active(self):
+        return db.session.query(Court).where(Court.status).all()
 
-    def separate_courts(self) -> Tuple[dict, dict]:
-        active_courts = {k: v for k, v in self.courts.items() if v}
-        inactive_courts = {k: v for k, v in self.courts.items() if not v}
-        return active_courts, inactive_courts
+    def all(self) -> list[Court]:
+        courts = [court for court in db.session.query(Court).order_by(Court.id)]
+        return courts
 
+    def reset(self):
+        for court in db.session.scalars(select(Court)):
+            court.status = True
+        db.session.commit()
 
-def _init_guests(session) -> None:
-    for level_id, level in enumerate(GUEST_LEVELS):
-        for player_id in range(GUESTS_PER_LEVEL):
-            unique_id = (level_id * GUESTS_PER_LEVEL) + player_id + 1
-            player = Player(
-                id=-unique_id,
-                first="Guest",
-                last=str(unique_id),
-                level=level,
-                status=False,
-                count=0,
-            )
-            session.add(player)
-    session.commit()
 
+class Player(db.Model):
+    id: Mapped[int] = mc(primary_key=True)
+    first_name: Mapped[str] = mc(String(64), index=True, default="---")
+    last_name: Mapped[str] = mc(String(64), index=True, default="---")
+    team: Mapped[int] = mc(Integer, index=True, default=0)
+    status: Mapped[bool] = mc(Boolean, default=False)
+    level: Mapped[int] = mc(Integer, default=app.config["MIN_PLAYER_LEVEL"])
 
-def _init_database_from_csv(session, csv_location: str) -> None:
-    data = read_csv(csv_location)
-    session.query(Player).delete()
-    for _, entry in data.iterrows():
-        player = Player(
-            id=entry["id"],
-            first=entry["first"],
-            last=entry["last"],
-            level=entry["skill"],
-            team=entry["team"],
-            status=False,
-            count=0,
-        )
-        session.add(player)
-    session.commit()
-    _init_guests(session)
+    def __repr__(self):
+        return f"{self.first_name} {self.last_name} is {"active" if self.status else "inactive"}"
 
 
-def _init_random_database(session, player_count: int = 80) -> None:
+def initialize_random_database(
+    player_count: int = 80, guest_per_level: int = 4
+) -> None:
+    db.session.add(Player(*{"first_name": "---", "last_name": ""}))
+    db.session.add(Player(*{"first_name": "RESERVED", "last_name": ""}))
     total_teams = 5
     players_per_team = 5
     for i in range(player_count):
@@ -85,88 +89,117 @@ def _init_random_database(session, player_count: int = 80) -> None:
             team = floor(i / players_per_team) + 1
         else:
             team = 0
-        player = Player(
-            id=i,
-            first=get_first_name(),
-            last=get_last_name(),
-            level=randint(MIN_LEVEL, MAX_LEVEL),
-            team=team,
-            status=bool(getrandbits(1)),
-            count=0,
-        )
-        session.add(player)
-    session.commit()
-    _init_guests(session)
+        args = {
+            "first_name": get_first_name(),
+            "last_name": get_last_name(),
+            "team": team,
+            "status": bool(getrandbits(1)),
+            "level": randint(
+                app.config["MIN_PLAYER_LEVEL"], app.config["MAX_PLAYER_LEVEL"]
+            ),
+        }
+        db.session.add(Player(*args))
+    meta_data = {"first_name": "Guest", "team": 0, "status": False}
+    for level_id, level in enumerate(app.config["GUEST_LEVELS"]):
+        for player_id in range(guest_per_level):
+            unique_id = (level_id * guest_per_level) + player_id + 1
+            args = meta_data | {"last_name": str(unique_id), "level": level}
+            db.session.add(Player(*args))
+    db.session.commit()
 
 
-class PlayerList:
+class Players:
 
-    def __init__(self, csv_location=None) -> None:
-        self.db_location = "players.db"
-        db_file_found = True
-        if not isfile(self.db_location):
-            db_file_found = False
-        self._init_database()
-        if csv_location is not None:
-            _init_database_from_csv(self.session, csv_location)
+    def __init__(self) -> None:
+        if app.config["DATABASE_CSV"] is not None:
+            data = read_csv(app.config["DATABASE_CSV"])
+            db.session.query(Player).delete()
+            for entry in data.to_dict("records"):
+                args = entry | {"status": False}
+                db.session.add(Player(*args))
+            db.session.commit()
         else:
-            if not db_file_found:
-                _init_random_database(self.session)
+            if len(self.all()) == 0:
+                initialize_random_database()
 
-    def _init_database(self):
-        engine = create_engine("sqlite:///" + str(self.db_location))
-        BASE.metadata.create_all(bind=engine)
-        self.session = sessionmaker(bind=engine)()
+    def default_id(self):
+        query = select(Player).where(Player.first_name == "---")
+        player = db.session.scalars(query).first()
+        if player is None:
+            raise ValueError(f"Database isn't initialized.")
+        return player.id
 
-    def get_player_status(self, id: int) -> bool:
-        return self.session.query(Player).filter(Player.id == id).first().status
+    def reserved_id(self):
+        query = select(Player).where(Player.first_name == "RESERVED")
+        player = db.session.scalars(query).first()
+        if player is None:
+            raise ValueError(f"Database isn't initialized.")
+        return player.id
 
-    def toggle_player_status(self, id: int) -> DisplayPlayer:
-        player = self.session.query(Player).filter(Player.id == id).first()
+    def get(self, id: int) -> bool:
+        player = db.session.get(Player, id)
+        if player is None:
+            raise ValueError(f"User {id} doesn't exist.")
+        return player.status
+
+    def toggle(self, id: int) -> Player:
+        player = db.session.get(Player, id)
+        if player is None:
+            raise ValueError(f"User {id} doesn't exist.")
         player.status = not player.status
-        self.session.commit()
-        return DisplayPlayer(player)
+        db.session.commit()
+        return player
 
-    def get_all_players(self) -> Tuple[list[DisplayPlayer], list[DisplayPlayer]]:
-        all_players = [
-            DisplayPlayer(p)
-            for p in self.session.query(Player)
-            .order_by(Player.first, Player.last)
-            .all()
+    def all(self) -> Tuple[list[Player], list[Player]]:
+        regulars = [
+            p for p in db.session.query(Player).where(Player.first_name != "Guest")
+        ]
+        guests = [
+            p for p in db.session.query(Player).where(Player.first_name == "Guest")
         ]
-        regulars = [player for player in all_players if player.first != "Guest"]
-        guests = [player for player in all_players if player.first == "Guest"]
-        guests.sort(key=lambda x: int(x.last))
+        guests.sort(key=lambda x: int(x.last_name))
         return regulars, guests
 
-    def get_possible_game_count(self) -> int:
-        total_players = len(self.get_players())
-        possible_games = floor(total_players / PLAYER_PER_COURT)
-        return possible_games
-
-    def get_players(self) -> list[Player]:
-        active_players = (
-            self.session.query(Player)
-            .order_by(Player.first, Player.last)
-            .filter(not_(Player.status == 0))
-            .all()
+    def active(self) -> Sequence[Player]:
+        query = (
+            select(Player)
+            .where(Player.status)
+            .order_by(Player.first_name, Player.last_name)
         )
-        return active_players
+        return db.session.scalars(query).all()
+
+    def reset(self):
+        for player in db.session.scalars(select(Player)):
+            player.status = False
+        db.session.commit()
+
+
+class Game(db.Model):
+    id: Mapped[int] = mc(primary_key=True)
+    round_id: Mapped[int] = mc(Integer, index=True)
+    court_id: Mapped[int] = mc(ForeignKey(Court.id), index=True)
+    player_1: Mapped[int] = mc(ForeignKey(Player.id), index=True)
+    player_2: Mapped[int] = mc(ForeignKey(Player.id), index=True)
+    player_3: Mapped[int] = mc(ForeignKey(Player.id), index=True)
+    player_4: Mapped[int] = mc(ForeignKey(Player.id), index=True)
+    date: Mapped[datetime] = mc(index=True, default=lambda: date.today())
+
 
-    def increment_game_count(self, player_ids: list[int]):
-        players = self.session.query(Player).filter(Player.id.in_(player_ids))
-        for player in players:
-            player.count += 1
-        self.session.commit()
+class DisplayPlayer:
 
-    def reset_game_count(self):
-        players = self.session.query(Player)
-        for player in players:
-            player.count = 0
-        self.session.commit()
+    def __init__(self, player: Player):
+        self.id = player.id
+        self.name = f"{player.first_name} {player.last_name}"
 
-    def reset_all_players(self):
-        players = self.session.query(Player)
-        for player in players:
-            player.status = 0
-        self.session.commit()
+
+@dataclass
+class DisplayTeam:
+    player_1: DisplayPlayer
+    player_2: DisplayPlayer
+
+
+@dataclass
+class DisplayGame:
+    court_id: int
+    team_1: DisplayTeam
+    team_2: DisplayTeam