diff options
| -rw-r--r-- | app/__init__.py | 4 | ||||
| -rw-r--r-- | app/config.py | 11 | ||||
| -rw-r--r-- | app/controller.py | 178 | ||||
| -rw-r--r-- | app/data.py | 76 | ||||
| -rw-r--r-- | app/model.py | 295 | ||||
| -rw-r--r-- | app/routes.py | 28 | ||||
| -rw-r--r-- | app/utilities.py | 106 | ||||
| -rw-r--r-- | migrations/README | 1 | ||||
| -rw-r--r-- | migrations/alembic.ini | 50 | ||||
| -rw-r--r-- | migrations/env.py | 113 | ||||
| -rw-r--r-- | migrations/script.py.mako | 24 | ||||
| -rw-r--r-- | migrations/versions/d158cf8f4950_.py | 39 |
12 files changed, 558 insertions, 367 deletions
diff --git a/app/__init__.py b/app/__init__.py index 7b0c03d..2e0a8be 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,7 +1,11 @@ from flask import Flask from app.config import Config +from flask_migrate import Migrate +from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) app.config.from_object(Config) +db = SQLAlchemy(app) +migrate = Migrate(app, db) from app import routes diff --git a/app/config.py b/app/config.py index aeb2840..a6a60be 100644 --- a/app/config.py +++ b/app/config.py @@ -1,4 +1,13 @@ from os import environ +from os.path import abspath, dirname, join + class Config: - SECRET_KEY = environ.get("SECRET_KEY") or "you-will-never-guess"
\ No newline at end of file + SECRET_KEY = environ.get("SECRET_KEY") or None + DATABASE_CSV = environ.get("DATABASE_CSV") or None + COURT_COUNT = environ.get("COURT_COUNT") or None + SQLALCHEMY_DATABASE_URI = "sqlite:///" + join(abspath(dirname(__file__)), "app.db") + PLAYERS_PER_COURT = 4 + GUEST_LEVELS = [1, 3, 5] + MIN_PLAYER_LEVEL = 1 + MAX_PLAYER_LEVEL = 10
\ No newline at end of file diff --git a/app/controller.py b/app/controller.py index 498715d..1937af8 100644 --- a/app/controller.py +++ b/app/controller.py @@ -1,99 +1,103 @@ -from typing import Tuple +from math import floor +from app import app, db +from datetime import date from operator import attrgetter -from app.data import Game, PLAYER_PER_COURT -from app.model import Player, PlayerList, CourtList +from sqlalchemy import select, func +from app.model import ( + Player, + Players, + Courts, + Game, + DisplayPlayer, + DisplayTeam, + DisplayGame, +) from app.utilities import ( - _create_placeholders, - _select_players, - _create_shuffle_games, - _create_team_games, + get_placeholder_games, + get_team_games, + get_shuffle_games, separate_teams_and_players, - split_courts, ) -class RoundGenerator: - proposed_games: list[Game] = [] - proposed_players: list[Player] = [] - - def __init__(self) -> None: - self.courts = CourtList() - self.players = PlayerList() - self.reset() - - def _separate_courts(self, courts: dict) -> Tuple[dict, dict]: - possible_games = min(len(courts), self.players.get_possible_game_count()) - courts_as_list = list(courts.items()) - used_courts = dict(courts_as_list[:possible_games]) - unused_courts = dict(courts_as_list[possible_games:]) - return used_courts, unused_courts +class Session: + levels: int = 10 + teams: bool = False + proposal: list[DisplayGame] = [] + courts: Courts = Courts() + players: Players = Players() - def _create_inactive_games(self) -> list[Game]: - active_courts, reserved_courts = self.courts.separate_courts() - reserved_games = _create_placeholders(reserved_courts, "RESERVED") - _, unused_courts = self._separate_courts(active_courts) - blank_games = _create_placeholders(unused_courts, "---") - games = reserved_games + blank_games - return games + def __init__(self): + games = db.session.query(Game).filter(Game.date.date() == date.today()).all() + if len(games) > 0: + return + id = self.players.default_id() + player_data = {f"player_{i+1}": id for i in range(4)} + for court in self.courts.all(): + args = {"round_id": 0, "court_id": court.id} | player_data + db.session.add(Game(*args)) + db.session.commit() - def _create_active_games( - self, total_levels: int, group_by_team: bool - ) -> Tuple[list[Game], list[Player]]: - active_courts = self.courts.separate_courts()[0] - used_courts = self._separate_courts(active_courts)[0] - required_player_count = len(used_courts) * PLAYER_PER_COURT - active_players = self.players.get_players() - if group_by_team: - active_teams, team_less_players = separate_teams_and_players(active_players) - regular_court_count = len(used_courts) - len(active_teams) - regular_courts, team_courts = split_courts(used_courts, regular_court_count) - new_player_count = required_player_count - ( - PLAYER_PER_COURT * len(active_teams) - ) - regular_players = _select_players(team_less_players, new_player_count) - regular_games = _create_shuffle_games( - regular_courts, regular_players, total_levels - ) - team_games, selected_team_players = _create_team_games( - team_courts, active_teams - ) - selected_players = regular_players + selected_team_players - games = regular_games + team_games - else: - selected_players = _select_players(active_players, required_player_count) - games = _create_shuffle_games(used_courts, selected_players, total_levels) - return games, selected_players + def round(self) -> int: + query = select(func.max(Game.round_id)).where(Game.date.date() == date.today()) + round_id = db.session.scalars(query).first() + if round_id is None: + raise ValueError(f"Database isn't initialized") + return round_id - def update_player_list(self, csv_location): - self.players = PlayerList(csv_location) - - def clear(self) -> list[Game]: - self.proposed_games = [] - self.proposed_players = [] - return self.games - - def propose(self, levels: int, group_by_team: bool) -> list[Game]: - inactive_games = self._create_inactive_games() - active_games, selected_players = self._create_active_games( - levels, group_by_team + def all(self) -> list[DisplayGame]: + games = ( + db.session.query(Game) + .filter(Game.date.date() == date.today()) + .filter(Game.round_id == self.round()) + .order_by(Game.court_id) + .all() ) - games = active_games + inactive_games - games.sort(key=attrgetter("court")) - self.proposed_games = games - self.proposed_players = selected_players - return self.proposed_games + dg = [] + for game in games: + dp = [] + for i in range(4): + p = db.session.get(Player, getattr(game, f"player_{i+1}")) + if p is None: + raise ValueError(f"Player id mismatch.") + t1 = DisplayTeam(DisplayPlayer(dp[0]), DisplayPlayer(dp[1])) + t2 = DisplayTeam(DisplayPlayer(dp[2]), DisplayPlayer(dp[3])) + db.session.get(Player, game.player_2) + dg.append(DisplayGame(game.court_id, t1, t2)) + return dg + + def propose(self) -> list[DisplayGame]: + reserved_courts = [court for court in self.courts.all() if not court.status] + reserved_games = get_placeholder_games(reserved_courts, "RESERVED") + active_courts = self.courts.active() + active_players = self.players.active() + game_count = floor(len(active_players) / app.config["PLAYER_PER_COURT"]) + game_count = min(len(active_courts), game_count) + used_courts = active_courts[:game_count] + unused_courts = active_courts[game_count:] + blank_games = get_placeholder_games(unused_courts, "---") + active_teams = {} + if self.teams: + active_teams, active_players = separate_teams_and_players(active_players) + team_courts = used_courts[: len(active_teams)] + shuffle_courts = used_courts[len(active_teams) :] + team_games = get_team_games(team_courts, active_teams) + shuffle_games = get_shuffle_games(shuffle_courts, active_players, self.levels) + self.proposal = shuffle_games + team_games + blank_games + reserved_games + self.proposal.sort(key=attrgetter("court_id")) + return self.proposal - def confirm(self) -> list[Game]: - if len(self.proposed_games) == 0: - return self.games - self.games = self.proposed_games - identifiers = [int(player.id) for player in self.proposed_players] - self.players.increment_game_count(identifiers) - self.round += 1 - return self.clear() + def confirm(self) -> list[DisplayGame]: + if len(self.proposal) == 0: + return self.all() + for game in self.proposal: + db.session.add(game) + db.session.commit() + self.proposal = [] + return self.all() - def reset(self): - self.round = 0 - self.games = _create_placeholders(self.courts.get_courts(), "---") - self.players.reset_game_count() - return self.games + def reset(self) -> list[DisplayGame]: + db.session.query(Game).filter(Game.date.date() == date.today()).delete() + db.session.commit() + self.__init__() + return self.all() diff --git a/app/data.py b/app/data.py deleted file mode 100644 index c00dde2..0000000 --- a/app/data.py +++ /dev/null @@ -1,76 +0,0 @@ -from dataclasses import dataclass -from sqlalchemy import Column, String, Integer -from sqlalchemy.ext.declarative import declarative_base - - -MIN_LEVEL = 1 -MAX_LEVEL = 10 -PLAYER_PER_COURT = 4 -GUEST_LEVELS = [1, 3, 5] -GUESTS_PER_LEVEL = 4 - -BASE = declarative_base() - - -class Player(BASE): - __tablename__ = "players" - id = Column("id", Integer, primary_key=True) - first = Column("first", String) - last = Column("last", String) - level = Column("level", Integer) - team = Column("team", Integer) - status = Column("status", Integer) - count = Column("count", Integer) - - def __init__( - self, - id: int, - first: str = "", - last: str = "", - level: int = 0, - team: int = 0, - status: bool = True, - count: int = 0, - ): - self.id = id - self.first = first - self.last = last - self.status = status - self.team = team - self.level = level - self.count = count - - def __repr__(self): - team = f" of Team {self.team}" if self.team else "" - status = "active" if self.status else "inactive" - return f"{self.first} {self.last} (Level {self.level}){team} is {status} and has played {self.count} games" - - -class DisplayPlayer: - id: int - first: str - last: str - status: bool - - def __init__(self, player: Player): - self.id = int(player.id) - self.first = str(player.first) - self.last = str(player.last) - self.status = bool(player.status) - - def __repr__(self): - status = "active" if self.status else "inactive" - return f"{self.first} {self.last} is {status}" - - -@dataclass -class Team: - player1: DisplayPlayer - player2: DisplayPlayer - - -@dataclass -class Game: - court: int - team1: Team - team2: Team 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 diff --git a/app/routes.py b/app/routes.py index b943e16..2cc6009 100644 --- a/app/routes.py +++ b/app/routes.py @@ -1,9 +1,9 @@ from app import app from app.forms import LoginForm -from app.controller import RoundGenerator +from app.controller import Session from flask import render_template, request, url_for, Response, flash, redirect -rg = RoundGenerator() +g = Session() @app.route("/", methods=["GET"]) def home(): @@ -19,50 +19,50 @@ def login(): @app.route("/control", methods=["GET"]) def get_controls(): - return render_template("controls.j2", courts=rg.courts.get_courts()) + return render_template("controls.j2", courts=g.courts.all()) @app.route("/propose", methods=["POST"]) def get_random_games(): - levels = int(request.form["levels"].split()[0]) - team_practice = request.form["team"] == "Teams" + g.levels = int(request.form["levels"].split()[0]) + g.teams = request.form["team"] == "Teams" return render_template( - "games.j2", games=rg.propose(levels, team_practice), title="Proposal" + "games.j2", games=g.propose(), title="Proposal" ) @app.route("/confirm", methods=["POST"]) def confirm_games(): - return render_template("games.j2", games=rg.confirm(), title=f"Round: {rg.round}") + return render_template("games.j2", games=g.confirm(), title=f"Round: {g.round()}") @app.route("/clear", methods=["POST"]) def clear_games(): - return render_template("games.j2", games=rg.clear(), title=f"Round: {rg.round}") + return render_template("games.j2", games=g.all(), title=f"Round: {g.round()}") @app.route("/reset", methods=["POST"]) def reset_games(): - return render_template("games.j2", games=rg.reset(), title=f"Round: {rg.round}") + return render_template("games.j2", games=g.reset(), title=f"Round: {g.round()}") @app.route("/player-toggle/<player_request>", methods=["POST"]) def toggle_player(player_request): id = int(player_request) - player = rg.players.toggle_player_status(id) + player = g.players.toggle(id) t = '{% from "macros.j2" import player_button %}{{player_button(player)}}' return render_template(t, player=player) @app.route("/players", methods=["GET"]) def get_list_of_players(): - regulars, guests = rg.players.get_all_players() + regulars, guests = g.players.all() return render_template("players.j2", regulars=regulars, guests=guests, url_for=url_for) @app.route("/reset_players", methods=["POST"]) def reset_all_players(): - rg.players.reset_all_players() + g.players.reset() response = Response("Players reset!") response.headers["HX-Refresh"] = "true" return response @@ -71,11 +71,11 @@ def reset_all_players(): @app.route("/court-toggle/<court_number>", methods=["POST"]) def toggle_court(court_number): id = int(court_number) - status = rg.courts.toggle_court_status(id) + status = g.courts.toggle(id) t = '{% from "macros.j2" import court_button %}{{ court_button(id, status) }}' return render_template(t, id=id, status=status) @app.route("/courts", methods=["GET"]) def get_list_of_courts(): - return render_template("courts.j2", courts=rg.courts.get_courts()) + return render_template("courts.j2", courts=g.courts.all()) diff --git a/app/utilities.py b/app/utilities.py index e324161..00dc14a 100644 --- a/app/utilities.py +++ b/app/utilities.py @@ -1,7 +1,7 @@ -from typing import Tuple +from app import app, db +from typing import Tuple, Sequence from random import shuffle, sample -from app.data import Player, DisplayPlayer, Team, Game -from app.data import MAX_LEVEL, PLAYER_PER_COURT +from app.model import Player, DisplayPlayer, Court, DisplayGame, DisplayTeam def _shuffle_two_lists(a: list, b: list) -> Tuple[list, list]: @@ -25,76 +25,34 @@ def sample_list(indices: list, cost: list[int], count=None) -> list: return indices -def _create_placeholders(courts: dict, name: str) -> list[Game]: - dummy_player = Player(id=-1, first=name) - dummy_team = Team( - player1=DisplayPlayer(dummy_player), player2=DisplayPlayer(dummy_player) - ) - games = [Game(court, dummy_team, dummy_team) for court in courts] - return games - - def _normalize_skill(level: int, new_max: int) -> int: - new_level = round((new_max - 1) * (level / MAX_LEVEL)) + 1 + new_level = round((new_max - 1) * (level / app.config["MAX_PLAYER_LEVEL"])) + 1 return new_level -def _select_players(players: list[Player], count: int) -> list[Player]: +def select_players(players: Sequence[Player], count: int) -> list[Player]: identifiers = [player.id for player in players] - play_count = [int(player.count) for player in players] + play_count = [int(player.id) for player in players] ids = sample_list(identifiers, play_count, count) selected_players = [player for player in players if player.id in ids] return selected_players -def _create_groups_of_four_players( - players: list[Player], new_max: int -) -> list[list[Player]]: +def group_by_four(players: list[Player], new_max: int) -> list[list[Player]]: identifiers = [player.id for player in players] player_levels = [_normalize_skill(int(player.level), new_max) for player in players] identifiers = sample_list(identifiers, player_levels) court_players = [] - for i in range(0, len(players), PLAYER_PER_COURT): - court_ids = identifiers[i : i + PLAYER_PER_COURT] + for i in range(0, len(players), app.config["PLAYERS_PER_COURT"]): + court_ids = identifiers[i : i + app.config["PLAYERS_PER_COURT"]] selected_players = [player for player in players if player.id in court_ids] court_players.append(selected_players) return court_players -def _create_shuffle_games( - courts: dict, players: list[Player], levels: int -) -> list[Game]: - court_players = _create_groups_of_four_players(players, levels) - games = [] - for court, players in zip(courts, court_players): - team1 = Team( - player1=DisplayPlayer(players[0]), player2=DisplayPlayer(players[3]) - ) - team2 = Team( - player1=DisplayPlayer(players[1]), player2=DisplayPlayer(players[2]) - ) - games.append(Game(court, team1, team2)) - return games - - -def _create_team_games(courts: dict, teams: dict) -> tuple[list[Game], list[Player]]: - games = [] - selected_players = [] - for court, (_, players) in zip(courts, teams.items()): - players = sample(players, PLAYER_PER_COURT) - selected_players += players - players.sort(key=lambda x: x.level, reverse=True) - team1 = Team( - player1=DisplayPlayer(players[0]), player2=DisplayPlayer(players[3]) - ) - team2 = Team( - player1=DisplayPlayer(players[1]), player2=DisplayPlayer(players[2]) - ) - games.append(Game(court, team1, team2)) - return games, selected_players - - -def separate_teams_and_players(players: list[Player]) -> Tuple[dict, list[Player]]: +def separate_teams_and_players( + players: Sequence[Player], +) -> tuple[dict[int, list[Player]], list[Player]]: unique_team_id = {player.team for player in players} grouped_players_by_team = {team_id: [] for team_id in unique_team_id} for player in players: @@ -117,7 +75,39 @@ def separate_teams_and_players(players: list[Player]) -> Tuple[dict, list[Player return active_teams, players_without_team -def split_courts(courts: dict, n) -> Tuple[dict, dict]: - first_half = {k: v for (k, v) in [x for x in courts.items()][:n]} - second_half = {k: v for (k, v) in [x for x in courts.items()][n:]} - return first_half, second_half +def get_placeholder_games(courts: list[Court], name: str) -> list[DisplayGame]: + p = db.session.query(Player).where(Player.first_name == name).first() + if p is None: + raise ValueError(f"Database isn't initialized.") + dummy_team = DisplayTeam(DisplayPlayer(p), DisplayPlayer(p)) + games = [DisplayGame(court.id, dummy_team, dummy_team) for court in courts] + return games + + +def get_team_games( + courts: list[Court], teams: dict[int, list[Player]] +) -> list[DisplayGame]: + games = [] + for court, (_, players) in zip(courts, teams.items()): + players = sample(players, app.config["PLAYER_PER_COURT"]) + players.sort(key=lambda x: x.level, reverse=True) + dp = [DisplayPlayer(p) for p in players] + games.append( + DisplayGame(court.id, DisplayTeam(dp[0], dp[3]), DisplayTeam(dp[1], dp[2])) + ) + return games + + +def get_shuffle_games( + courts: list[Court], players: Sequence[Player], total_levels: int +) -> list[DisplayGame]: + required_player_count = len(courts) * app.config["PLAYER_PER_COURT"] + selected_players = select_players(players, required_player_count) + court_players = group_by_four(selected_players, total_levels) + games = [] + for court, players in zip(courts, court_players): + dp = [DisplayPlayer(p) for p in players] + games.append( + DisplayGame(court.id, DisplayTeam(dp[0], dp[3]), DisplayTeam(dp[1], dp[2])) + ) + return games diff --git a/migrations/README b/migrations/README new file mode 100644 index 0000000..0e04844 --- /dev/null +++ b/migrations/README @@ -0,0 +1 @@ +Single-database configuration for Flask. diff --git a/migrations/alembic.ini b/migrations/alembic.ini new file mode 100644 index 0000000..ec9d45c --- /dev/null +++ b/migrations/alembic.ini @@ -0,0 +1,50 @@ +# A generic, single database configuration. + +[alembic] +# template used to generate migration files +# file_template = %%(rev)s_%%(slug)s + +# set to 'true' to run the environment during +# the 'revision' command, regardless of autogenerate +# revision_environment = false + + +# Logging configuration +[loggers] +keys = root,sqlalchemy,alembic,flask_migrate + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[logger_flask_migrate] +level = INFO +handlers = +qualname = flask_migrate + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/migrations/env.py b/migrations/env.py new file mode 100644 index 0000000..4c97092 --- /dev/null +++ b/migrations/env.py @@ -0,0 +1,113 @@ +import logging +from logging.config import fileConfig + +from flask import current_app + +from alembic import context + +# this is the Alembic Config object, which provides +# access to the values within the .ini file in use. +config = context.config + +# Interpret the config file for Python logging. +# This line sets up loggers basically. +fileConfig(config.config_file_name) +logger = logging.getLogger('alembic.env') + + +def get_engine(): + try: + # this works with Flask-SQLAlchemy<3 and Alchemical + return current_app.extensions['migrate'].db.get_engine() + except (TypeError, AttributeError): + # this works with Flask-SQLAlchemy>=3 + return current_app.extensions['migrate'].db.engine + + +def get_engine_url(): + try: + return get_engine().url.render_as_string(hide_password=False).replace( + '%', '%%') + except AttributeError: + return str(get_engine().url).replace('%', '%%') + + +# add your model's MetaData object here +# for 'autogenerate' support +# from myapp import mymodel +# target_metadata = mymodel.Base.metadata +config.set_main_option('sqlalchemy.url', get_engine_url()) +target_db = current_app.extensions['migrate'].db + +# other values from the config, defined by the needs of env.py, +# can be acquired: +# my_important_option = config.get_main_option("my_important_option") +# ... etc. + + +def get_metadata(): + if hasattr(target_db, 'metadatas'): + return target_db.metadatas[None] + return target_db.metadata + + +def run_migrations_offline(): + """Run migrations in 'offline' mode. + + This configures the context with just a URL + and not an Engine, though an Engine is acceptable + here as well. By skipping the Engine creation + we don't even need a DBAPI to be available. + + Calls to context.execute() here emit the given string to the + script output. + + """ + url = config.get_main_option("sqlalchemy.url") + context.configure( + url=url, target_metadata=get_metadata(), literal_binds=True + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online(): + """Run migrations in 'online' mode. + + In this scenario we need to create an Engine + and associate a connection with the context. + + """ + + # this callback is used to prevent an auto-migration from being generated + # when there are no changes to the schema + # reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html + def process_revision_directives(context, revision, directives): + if getattr(config.cmd_opts, 'autogenerate', False): + script = directives[0] + if script.upgrade_ops.is_empty(): + directives[:] = [] + logger.info('No changes in schema detected.') + + conf_args = current_app.extensions['migrate'].configure_args + if conf_args.get("process_revision_directives") is None: + conf_args["process_revision_directives"] = process_revision_directives + + connectable = get_engine() + + with connectable.connect() as connection: + context.configure( + connection=connection, + target_metadata=get_metadata(), + **conf_args + ) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/migrations/script.py.mako b/migrations/script.py.mako new file mode 100644 index 0000000..2c01563 --- /dev/null +++ b/migrations/script.py.mako @@ -0,0 +1,24 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision = ${repr(up_revision)} +down_revision = ${repr(down_revision)} +branch_labels = ${repr(branch_labels)} +depends_on = ${repr(depends_on)} + + +def upgrade(): + ${upgrades if upgrades else "pass"} + + +def downgrade(): + ${downgrades if downgrades else "pass"} diff --git a/migrations/versions/d158cf8f4950_.py b/migrations/versions/d158cf8f4950_.py new file mode 100644 index 0000000..392d3bf --- /dev/null +++ b/migrations/versions/d158cf8f4950_.py @@ -0,0 +1,39 @@ +"""empty message + +Revision ID: d158cf8f4950 +Revises: +Create Date: 2025-07-06 15:31:24.124657 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'd158cf8f4950' +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('user', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('username', sa.String(length=64), nullable=False), + sa.Column('password_hash', sa.String(length=256), nullable=True), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('user', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_user_username'), ['username'], unique=True) + + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('user', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_user_username')) + + op.drop_table('user') + # ### end Alembic commands ### |