christoph/flask-formula10
1
Fork 0
Code Issues 14 Actions Packages Activity

Large database migration
All checks were successful
Build Formula10 Docker Image / build-docker (push) Successful in 15s
Details

Browse Source
This commit is contained in:
Christoph Urlacher
2024-03-03 15:38:35 +01:00
parent 96cb8ca891
commit d3097038a5
34 changed files with 307 additions and 593 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
formula10/database/common_queries.py
View File

@ -3,8 +3,8 @@ from formula10.database.model.db_race_result import DbRaceResult
from formula10.database.model.db_user import DbUser
from formula10 import db
def race_has_result(race_name: str) -> bool:
return db.session.query(DbRaceResult).filter_by(race_name=race_name).first() is not None
def race_has_result(race_id: int) -> bool:
return db.session.query(DbRaceResult).filter_by(race_id=race_id).first() is not None
def user_exists_and_enabled(user_name: str) -> bool:
@ -15,9 +15,9 @@ def user_exists_and_disabled(user_name: str) -> bool:
return db.session.query(DbUser).filter_by(name=user_name, enabled=False).first() is not None
def find_single_driver_strict(driver_name: str) -> DbDriver:
db_driver: DbDriver | None = db.session.query(DbDriver).filter_by(name=driver_name).first()
def find_single_driver_strict(driver_id: int) -> DbDriver:
db_driver: DbDriver | None = db.session.query(DbDriver).filter_by(id=driver_id).first()
if db_driver is None:
raise Exception(f"Could not find driver with name {driver_name} in database")
raise Exception(f"Could not find driver with id {driver_id} in database")
return db_driver

111
formula10/database/import_export.py
View File

@ -1,111 +0,0 @@
import csv
import os.path
from typing import List, Any
from formula10 import db
from formula10.database.model.db_driver import DbDriver
from formula10.database.model.db_race import DbRace
from formula10.database.model.db_race_guess import DbRaceGuess
from formula10.database.model.db_race_result import DbRaceResult
from formula10.database.model.db_season_guess import DbSeasonGuess
from formula10.database.model.db_season_guess_result import DbSeasonGuessResult
from formula10.database.model.db_team import DbTeam
from formula10.database.model.db_user import DbUser
def load_csv(filename: str) -> List[List[str]]:
if not os.path.exists(filename):
print(f"Could not load data from file {filename}, as it doesn't exist!")
return []
with open(filename, "r", newline="") as file:
reader = csv.reader(file, delimiter=",")
next(reader, None) # skip header
return list(reader)
def write_csv(filename: str, objects: List[Any]):
if len(objects) == 0:
print(f"Could not write objects to file {filename}, as no objects were given!")
return
with open(filename, "w", newline="") as file:
writer = csv.writer(file, delimiter=",")
writer.writerow(objects[0].__csv_header__)
for obj in objects:
writer.writerow(obj.to_csv())
# Reload static database data, this has to be called from the app context
def reload_static_data():
print("Initializing database with static values...")
# Create it/update tables (if it/they doesn't exist!)
db.create_all()
# Clear static data
db.session.query(DbTeam).delete()
db.session.query(DbDriver).delete()
db.session.query(DbRace).delete()
# Reload static data
for row in load_csv("data/static_import/teams.csv"):
db.session.add(DbTeam.from_csv(row))
for row in load_csv("data/static_import/drivers.csv"):
db.session.add(DbDriver.from_csv(row))
for row in load_csv("data/static_import/races.csv"):
db.session.add(DbRace.from_csv(row))
db.session.commit()
def reload_dynamic_data():
print("Initializing database with dynamic values...")
# Create it/update tables (if it/they doesn't exist!)
db.create_all()
# Clear dynamic data
db.session.query(DbUser).delete()
db.session.query(DbRaceResult).delete()
db.session.query(DbRaceGuess).delete()
db.session.query(DbSeasonGuess).delete()
# Reload dynamic data
for row in load_csv("data/dynamic_export/users.csv"):
db.session.add(DbUser.from_csv(row))
for row in load_csv("data/dynamic_export/raceresults.csv"):
db.session.add(DbRaceResult.from_csv(row))
for row in load_csv("data/dynamic_export/raceguesses.csv"):
db.session.add(DbRaceGuess.from_csv(row))
for row in load_csv("data/dynamic_export/seasonguesses.csv"):
db.session.add(DbSeasonGuess.from_csv(row))
db.session.commit()
def reload_season_guess_result_data():
print("Loading season guess results...")
# Create it/update tables (if it/they doesn't exist!)
db.create_all()
# Clear result data
db.session.query(DbSeasonGuessResult).delete()
# Reload result data
for row in load_csv("data/static_import/season_guess_results.csv"):
db.session.add(DbSeasonGuessResult.from_csv(row))
db.session.commit()
def export_dynamic_data():
print("Exporting Userdata...")
users: List[DbUser] = db.session.query(DbUser).all()
raceresults: List[DbRaceResult] = db.session.query(DbRaceResult).all()
raceguesses: List[DbRaceGuess] = db.session.query(DbRaceGuess).all()
seasonguesses: List[DbSeasonGuess] = db.session.query(DbSeasonGuess).all()
write_csv("data/dynamic_export/users.csv", users)
write_csv("data/dynamic_export/raceresults.csv", raceresults)
write_csv("data/dynamic_export/raceguesses.csv", raceguesses)
write_csv("data/dynamic_export/seasonguesses.csv", seasonguesses)

26
formula10/database/model/db_driver.py
View File

@ -1,5 +1,4 @@
from typing import List
from sqlalchemy import String, ForeignKey
from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import mapped_column, Mapped, relationship
from formula10.database.model.db_team import DbTeam
@ -13,21 +12,14 @@ class DbDriver(db.Model):
"""
__tablename__ = "driver"
def __init__(self, *, name: str):
self.name = name # Primary key
def __init__(self, *, id: int):
self.id = id # Primary key
@classmethod
def from_csv(cls, row: List[str]):
db_driver: DbDriver = cls(name=str(row[0]))
db_driver.abbr = str(row[1])
db_driver.team_name = str(row[2])
db_driver.country_code = str(row[3])
return db_driver
name: Mapped[str] = mapped_column(String(32), primary_key=True)
abbr: Mapped[str] = mapped_column(String(4))
team_name: Mapped[str] = mapped_column(ForeignKey("team.name"))
country_code: Mapped[str] = mapped_column(String(2)) # alpha-2 code
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=False)
name: Mapped[str] = mapped_column(String(32), nullable=False, unique=True)
abbr: Mapped[str] = mapped_column(String(4), nullable=False, unique=True)
team_id: Mapped[str] = mapped_column(ForeignKey("team.id"), nullable=False)
country_code: Mapped[str] = mapped_column(String(2), nullable=False) # alpha-2 code
# Relationships
team: Mapped[DbTeam] = relationship("DbTeam", foreign_keys=[team_name])
team: Mapped[DbTeam] = relationship("DbTeam", foreign_keys=[team_id])

26
formula10/database/model/db_race.py
View File

@ -1,5 +1,4 @@
from datetime import datetime
from typing import List
from sqlalchemy import DateTime, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
@ -13,22 +12,11 @@ class DbRace(db.Model):
"""
__tablename__ = "race"
def __init__(self, *, name: str, number: int, date: datetime, pxx: int):
self.name = name # Primary key
def __init__(self, *, id: int):
self.id = id # Primary key
self.number = number
self.date = date
self.pxx = pxx
@classmethod
def from_csv(cls, row: List[str]):
db_race: DbRace = cls(name=str(row[0]),
number=int(row[1]),
date=datetime.strptime(str(row[2]), "%Y-%m-%d-%H-%M"),
pxx=int(row[3]))
return db_race
name: Mapped[str] = mapped_column(String(64), primary_key=True)
number: Mapped[int] = mapped_column(Integer)
date: Mapped[datetime] = mapped_column(DateTime)
pxx: Mapped[int] = mapped_column(Integer) # This is the place to guess
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=False)
name: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
number: Mapped[int] = mapped_column(Integer, nullable=False, unique=True)
date: Mapped[datetime] = mapped_column(DateTime, nullable=False, unique=True)
pxx: Mapped[int] = mapped_column(Integer, nullable=False) # This is the place to guess

43
formula10/database/model/db_race_guess.py
View File

@ -1,4 +1,3 @@
from typing import Any, List
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
@ -14,38 +13,18 @@ class DbRaceGuess(db.Model):
It stores the corresponding race and the guessed drivers for PXX and DNF.
"""
__tablename__ = "raceguess"
__csv_header__ = ["user_name", "race_name", "pxx_driver_name", "dnf_driver_name"]
def __init__(self, *, user_name: str, race_name: str, pxx_driver_name: str, dnf_driver_name: str):
self.user_name = user_name # Primary key
self.race_name = race_name # Primary key
def __init__(self, *, user_id: int, race_id: int):
self.user_id = user_id # Primary key
self.race_id = race_id # Primary key
self.dnf_driver_name = dnf_driver_name
self.pxx_driver_name = pxx_driver_name
@classmethod
def from_csv(cls, row: List[str]):
db_race_guess: DbRaceGuess = cls(user_name=str(row[0]),
race_name=str(row[1]),
pxx_driver_name=str(row[2]),
dnf_driver_name=str(row[3]))
return db_race_guess
def to_csv(self) -> List[Any]:
return [
self.user_name,
self.race_name,
self.pxx_driver_name,
self.dnf_driver_name
]
user_name: Mapped[str] = mapped_column(ForeignKey("user.name"), primary_key=True)
race_name: Mapped[str] = mapped_column(ForeignKey("race.name"), primary_key=True)
pxx_driver_name: Mapped[str] = mapped_column(ForeignKey("driver.name"))
dnf_driver_name: Mapped[str] = mapped_column(ForeignKey("driver.name"))
user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
race_id: Mapped[int] = mapped_column(ForeignKey("race.id"), primary_key=True)
pxx_driver_id: Mapped[int] = mapped_column(ForeignKey("driver.id"), nullable=False)
dnf_driver_id: Mapped[int] = mapped_column(ForeignKey("driver.id"), nullable=False)
# Relationships
user: Mapped[DbUser] = relationship("DbUser", foreign_keys=[user_name])
race: Mapped[DbRace] = relationship("DbRace", foreign_keys=[race_name])
pxx: Mapped[DbDriver] = relationship("DbDriver", foreign_keys=[pxx_driver_name])
dnf: Mapped[DbDriver] = relationship("DbDriver", foreign_keys=[dnf_driver_name])
user: Mapped[DbUser] = relationship("DbUser", foreign_keys=[user_id])
race: Mapped[DbRace] = relationship("DbRace", foreign_keys=[race_id])
pxx: Mapped[DbDriver] = relationship("DbDriver", foreign_keys=[pxx_driver_id])
dnf: Mapped[DbDriver] = relationship("DbDriver", foreign_keys=[dnf_driver_id])

41
formula10/database/model/db_race_result.py
View File

@ -1,4 +1,3 @@
from typing import Any, List
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
@ -11,39 +10,15 @@ class DbRaceResult(db.Model):
It stores the corresponding race and dictionaries of place-/dnf-order and a list of drivers that are excluded from the standings for this race.
"""
__tablename__ = "raceresult"
__csv_header__ = ["race_name", "pxx_driver_names_json", "first_dnf_driver_names_json", "dnf_driver_names_json", "excluded_driver_names_json"]
def __init__(self, *, race_name: str, pxx_driver_names_json: str, first_dnf_driver_names_json: str, dnf_driver_names_json: str, excluded_driver_names_json: str):
self.race_name = race_name # Primary key
def __init__(self, *, race_id: int):
self.race_id = race_id # Primary key
self.pxx_driver_names_json = pxx_driver_names_json
self.first_dnf_driver_names_json = first_dnf_driver_names_json
self.dnf_driver_names_json = dnf_driver_names_json
self.excluded_driver_names_json = excluded_driver_names_json
@classmethod
def from_csv(cls, row: List[str]):
db_race_result: DbRaceResult = cls(race_name=str(row[0]),
pxx_driver_names_json=str(row[1]),
first_dnf_driver_names_json=str(row[2]),
dnf_driver_names_json=str(row[3]),
excluded_driver_names_json=str(row[4]))
return db_race_result
def to_csv(self) -> List[Any]:
return [
self.race_name,
self.pxx_driver_names_json,
self.first_dnf_driver_names_json,
self.dnf_driver_names_json,
self.excluded_driver_names_json
]
race_name: Mapped[str] = mapped_column(ForeignKey("race.name"), primary_key=True)
pxx_driver_names_json: Mapped[str] = mapped_column(String(1024))
first_dnf_driver_names_json: Mapped[str] = mapped_column(String(1024))
dnf_driver_names_json: Mapped[str] = mapped_column(String(1024))
excluded_driver_names_json: Mapped[str] = mapped_column(String(1024))
race_id: Mapped[int] = mapped_column(ForeignKey("race.id"), primary_key=True)
pxx_driver_ids_json: Mapped[str] = mapped_column(String(1024), nullable=False)
first_dnf_driver_ids_json: Mapped[str] = mapped_column(String(1024), nullable=False)
dnf_driver_ids_json: Mapped[str] = mapped_column(String(1024), nullable=False)
excluded_driver_ids_json: Mapped[str] = mapped_column(String(1024), nullable=False)
# Relationships
race: Mapped[DbRace] = relationship("DbRace", foreign_keys=[race_name])
race: Mapped[DbRace] = relationship("DbRace", foreign_keys=[race_id])

65
formula10/database/model/db_season_guess.py
View File

@ -1,4 +1,3 @@
from typing import Any, List
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
@ -12,56 +11,24 @@ class DbSeasonGuess(db.Model):
A collection of bonus guesses for the entire season.
"""
__tablename__ = "seasonguess"
__csv_header__ = ["user_name", "hot_take", "p2_team_name",
"overtake_driver_name", "dnf_driver_name", "gained_driver_name", "lost_driver_name",
"team_winners_driver_names_json", "podium_drivers_driver_names_json"]
def __init__(self, *, user_name: str, team_winners_driver_names_json: str, podium_drivers_driver_names_json: str):
self.user_name = user_name # Primary key
def __init__(self, *, user_id: int):
self.user_id = user_id # Primary key
self.team_winners_driver_names_json = team_winners_driver_names_json
self.podium_drivers_driver_names_json = podium_drivers_driver_names_json
@classmethod
def from_csv(cls, row: List[str]):
db_season_guess: DbSeasonGuess = cls(user_name=str(row[0]),
team_winners_driver_names_json=str(row[7]),
podium_drivers_driver_names_json=str(row[8]))
db_season_guess.hot_take = str(row[1])
db_season_guess.p2_team_name = str(row[2])
db_season_guess.overtake_driver_name = str(row[3])
db_season_guess.dnf_driver_name = str(row[4])
db_season_guess.gained_driver_name = str(row[5])
db_season_guess.lost_driver_name = str(row[6])
return db_season_guess
def to_csv(self) -> List[Any]:
return [
self.user_name,
self.hot_take,
self.p2_team_name,
self.overtake_driver_name,
self.dnf_driver_name,
self.gained_driver_name,
self.lost_driver_name,
self.team_winners_driver_names_json,
self.podium_drivers_driver_names_json
]
user_name: Mapped[str] = mapped_column(ForeignKey("user.name"), primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
hot_take: Mapped[str | None] = mapped_column(String(512), nullable=True)
p2_team_name: Mapped[str | None] = mapped_column(ForeignKey("team.name"), nullable=True)
overtake_driver_name: Mapped[str | None] = mapped_column(ForeignKey("driver.name"), nullable=True)
dnf_driver_name: Mapped[str | None] = mapped_column(ForeignKey("driver.name"), nullable=True)
gained_driver_name: Mapped[str | None] = mapped_column(ForeignKey("driver.name"), nullable=True)
lost_driver_name: Mapped[str | None] = mapped_column(ForeignKey("driver.name"), nullable=True)
team_winners_driver_names_json: Mapped[str] = mapped_column(String(1024))
podium_drivers_driver_names_json: Mapped[str] = mapped_column(String(1024))
p2_team_id: Mapped[int | None] = mapped_column(ForeignKey("team.id"), nullable=True)
overtake_driver_id: Mapped[int | None] = mapped_column(ForeignKey("driver.id"), nullable=True)
dnf_driver_id: Mapped[int | None] = mapped_column(ForeignKey("driver.id"), nullable=True)
gained_driver_id: Mapped[int | None] = mapped_column(ForeignKey("driver.id"), nullable=True)
lost_driver_id: Mapped[int | None] = mapped_column(ForeignKey("driver.id"), nullable=True)
team_winners_driver_ids_json: Mapped[str] = mapped_column(String(1024))
podium_drivers_driver_ids_json: Mapped[str] = mapped_column(String(1024))
# Relationships
user: Mapped[DbUser] = relationship("DbUser", foreign_keys=[user_name])
p2_team: Mapped[DbTeam | None] = relationship("DbTeam", foreign_keys=[p2_team_name])
overtake_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[overtake_driver_name])
dnf_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[dnf_driver_name])
gained_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[gained_driver_name])
lost_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[lost_driver_name])
user: Mapped[DbUser] = relationship("DbUser", foreign_keys=[user_id])
p2_team: Mapped[DbTeam | None] = relationship("DbTeam", foreign_keys=[p2_team_id])
overtake_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[overtake_driver_id])
dnf_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[dnf_driver_id])
gained_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[gained_driver_id])
lost_driver: Mapped[DbDriver | None] = relationship("DbDriver", foreign_keys=[lost_driver_id])

34
formula10/database/model/db_season_guess_result.py
View File

@ -1,5 +1,3 @@
from typing import List
from sqlalchemy import Boolean, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
@ -12,33 +10,13 @@ class DbSeasonGuessResult(db.Model):
"""
__tablename__ = "seasonguessresult"
__csv_header__ = ["user_name", "hot_take_correct", "overtakes_correct"]
def __init__(self, *, user_name: str, hot_take_correct: bool, overtakes_correct: bool):
self.user_name = user_name # Primary key
def __init__(self, *, user_id: int):
self.user_id = user_id # Primary key
self.hot_take_correct = hot_take_correct
self.overtakes_correct = overtakes_correct
@classmethod
def from_csv(cls, row: List[str]):
db_season_guess_result: DbSeasonGuessResult = cls(user_name=str(row[0]),
hot_take_correct=True if str(row[1])=="True" else False,
overtakes_correct=True if str(row[2])=="True" else False)
return db_season_guess_result
# This object can't be edited from the page context
# def to_csv(self) -> List[Any]:
# return [
# self.user_name,
# self.hot_take_correct,
# self.overtakes_correct
# ]
user_name: Mapped[str] = mapped_column(ForeignKey("user.name"), primary_key=True)
hot_take_correct: Mapped[bool] = mapped_column(Boolean)
overtakes_correct: Mapped[bool] = mapped_column(Boolean)
user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
hot_take_correct: Mapped[bool] = mapped_column(Boolean, nullable=False)
overtakes_correct: Mapped[bool] = mapped_column(Boolean, nullable=False)
# Relationships
user: Mapped[DbUser] = relationship("DbUser", foreign_keys=[user_name])
user: Mapped[DbUser] = relationship("DbUser", foreign_keys=[user_id])

15
formula10/database/model/db_team.py
View File

@ -1,5 +1,4 @@
from typing import List
from sqlalchemy import String
from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from formula10 import db
@ -11,12 +10,8 @@ class DbTeam(db.Model):
"""
__tablename__ = "team"
def __init__(self, *, name: str):
self.name = name # Primary key
def __init__(self, *, id: int):
self.id = id # Primary key
@classmethod
def from_csv(cls, row: List[str]):
db_team: DbTeam = cls(name=str(row[0]))
return db_team
name: Mapped[str] = mapped_column(String(32), primary_key=True)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=False)
name: Mapped[str] = mapped_column(String(32), nullable=False, unique=True)

28
formula10/database/model/db_user.py
View File

@ -1,5 +1,4 @@
from typing import Any, List
from sqlalchemy import Boolean, String
from sqlalchemy import Boolean, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
from formula10 import db
@ -10,24 +9,11 @@ class DbUser(db.Model):
A user that can guess races (name only).
"""
__tablename__ = "user"
__csv_header__ = ["name", "enabled"]
def __init__(self, *, name: str, enabled: bool):
self.name = name # Primary key
def __init__(self, *, id: int | None):
if id is not None:
self.id = id # Primary key
self.enabled = enabled
@classmethod
def from_csv(cls, row: List[str]):
db_user: DbUser = cls(name=str(row[0]),
enabled=True if str(row[1])=="True" else False)
return db_user
def to_csv(self) -> List[Any]:
return [
self.name,
self.enabled
]
name: Mapped[str] = mapped_column(String(32), primary_key=True)
enabled: Mapped[bool] = mapped_column(Boolean)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(32), nullable=False, unique=True)
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False)

131
formula10/database/update_queries.py
View File

@ -6,6 +6,7 @@ from werkzeug import Response
from formula10.controller.error_controller import error_redirect
from formula10.database.common_queries import race_has_result, user_exists_and_disabled, user_exists_and_enabled
from formula10.database.model.db_race import DbRace
from formula10.database.model.db_race_guess import DbRaceGuess
from formula10.database.model.db_race_result import DbRaceResult
from formula10.database.model.db_season_guess import DbSeasonGuess
@ -14,159 +15,167 @@ from formula10.database.validation import any_is_none, positions_are_contiguous,
from formula10 import ENABLE_TIMING, db
def find_or_create_race_guess(user_name: str, race_name: str) -> DbRaceGuess:
def find_or_create_race_guess(user_id: int, race_id: int) -> DbRaceGuess:
# There can be a single RaceGuess at most, since (user_name, race_name) is the composite primary key
race_guess: DbRaceGuess | None = db.session.query(DbRaceGuess).filter_by(user_name=user_name, race_name=race_name).first()
race_guess: DbRaceGuess | None = db.session.query(DbRaceGuess).filter_by(user_id=user_id, race_id=race_id).first()
if race_guess is not None:
return race_guess
# Insert a new RaceGuess
race_guess = DbRaceGuess(user_name=user_name, race_name=race_name, pxx_driver_name="TEMP", dnf_driver_name="TEMP")
race_guess = DbRaceGuess(user_id=user_id, race_id=race_id)
race_guess.pxx_driver_id = 9999
race_guess.dnf_driver_id = 9999
db.session.add(race_guess)
db.session.commit()
# Double check if database insertion worked and obtain any values set by the database
race_guess = db.session.query(DbRaceGuess).filter_by(user_name=user_name, race_name=race_name).first()
race_guess = db.session.query(DbRaceGuess).filter_by(user_id=user_id, race_id=race_id).first()
if race_guess is None:
raise Exception("Failed adding RaceGuess to the database")
return race_guess
def update_race_guess(race_name: str, user_name: str, pxx_select: str | None, dnf_select: str | None) -> Response:
if any_is_none(pxx_select, dnf_select):
return error_redirect(f"Picks for race \"{race_name}\" were not saved, because you did not fill all the fields.")
def update_race_guess(race_id: int, user_id: int, pxx_select_id: int | None, dnf_select_id: int | None) -> Response:
if any_is_none(pxx_select_id, dnf_select_id):
return error_redirect(f"Picks for race \"{race_id}\" were not saved, because you did not fill all the fields.")
if ENABLE_TIMING and race_has_started(race_name=race_name):
return error_redirect(f"No picks for race \"{race_name}\" can be entered, as this race has already started.")
if ENABLE_TIMING and race_has_started(race_id=race_id):
return error_redirect(f"No picks for race \"{race_id}\" can be entered, as this race has already started.")
if race_has_result(race_name):
return error_redirect(f"No picks for race \"{race_name}\" can be entered, as this race has already finished.")
if race_has_result(race_id):
return error_redirect(f"No picks for race \"{race_id}\" can be entered, as this race has already finished.")
pxx_driver_name: str = cast(str, pxx_select)
dnf_driver_name: str = cast(str, dnf_select)
pxx_driver_id: int = cast(int, pxx_select_id)
dnf_driver_id: int = cast(int, dnf_select_id)
race_guess: DbRaceGuess = find_or_create_race_guess(user_name, race_name)
race_guess.pxx_driver_name = pxx_driver_name
race_guess.dnf_driver_name = dnf_driver_name
race_guess: DbRaceGuess = find_or_create_race_guess(user_id, race_id)
race_guess.pxx_driver_id = pxx_driver_id
race_guess.dnf_driver_id = dnf_driver_id
db.session.commit()
return redirect("/race/Everyone")
def delete_race_guess(race_name: str, user_name: str) -> Response:
def delete_race_guess(race_id: int, user_id: int) -> Response:
# Don't change guesses that are already over
if ENABLE_TIMING and race_has_started(race_name=race_name):
return error_redirect(f"No picks for race \"{race_name}\" can be deleted, as this race has already started.")
if ENABLE_TIMING and race_has_started(race_id=race_id):
return error_redirect(f"No picks for race with id \"{race_id}\" can be deleted, as this race has already started.")
if race_has_result(race_name):
return error_redirect(f"No picks for race \"{race_name}\" can be deleted, as this race has already finished.")
if race_has_result(race_id):
return error_redirect(f"No picks for race \"{race_id}\" can be deleted, as this race has already finished.")
# Does not throw if row doesn't exist
db.session.query(DbRaceGuess).filter_by(race_name=race_name, user_name=user_name).delete()
db.session.query(DbRaceGuess).filter_by(race_id=race_id, user_id=user_id).delete()
db.session.commit()
return redirect("/race/Everyone")
def find_or_create_season_guess(user_name: str) -> DbSeasonGuess:
def find_or_create_season_guess(user_id: int) -> DbSeasonGuess:
# There can be a single SeasonGuess at most, since user_name is the primary key
season_guess: DbSeasonGuess | None = db.session.query(DbSeasonGuess).filter_by(user_name=user_name).first()
season_guess: DbSeasonGuess | None = db.session.query(DbSeasonGuess).filter_by(user_id=user_id).first()
if season_guess is not None:
return season_guess
# Insert a new SeasonGuess
season_guess = DbSeasonGuess(user_name=user_name, team_winners_driver_names_json=json.dumps(["TEMP"]), podium_drivers_driver_names_json=json.dumps(["TEMP"]))
season_guess = DbSeasonGuess(user_id=user_id)
season_guess.team_winners_driver_ids_json=json.dumps(["9999"])
season_guess.podium_drivers_driver_ids_json=json.dumps(["9999"])
db.session.add(season_guess)
db.session.commit()
# Double check if database insertion worked and obtain any values set by the database
season_guess = db.session.query(DbSeasonGuess).filter_by(user_name=user_name).first()
season_guess = db.session.query(DbSeasonGuess).filter_by(user_id=user_id).first()
if season_guess is None:
raise Exception("Failed adding SeasonGuess to the database")
return season_guess
def update_season_guess(user_name: str, guesses: List[str | None], team_winner_guesses: List[str | None], podium_driver_guesses: List[str]) -> Response:
def update_season_guess(user_id: int, guesses: List[str | None], team_winner_guesses: List[str | None], podium_driver_guesses: List[str]) -> Response:
# Pylance marks type errors here, but those are intended. Columns are marked nullable.
if ENABLE_TIMING and race_has_started(race_name="Bahrain"):
if ENABLE_TIMING and race_has_started(race_id=1):
return error_redirect("No season picks can be entered, as the season has already begun!")
season_guess: DbSeasonGuess = find_or_create_season_guess(user_name)
season_guess: DbSeasonGuess = find_or_create_season_guess(user_id)
season_guess.hot_take = guesses[0] # type: ignore
season_guess.p2_team_name = guesses[1] # type: ignore
season_guess.overtake_driver_name = guesses[2] # type: ignore
season_guess.dnf_driver_name = guesses[3] # type: ignore
season_guess.gained_driver_name = guesses[4] # type: ignore
season_guess.lost_driver_name = guesses[5] # type: ignore
season_guess.team_winners_driver_names_json = json.dumps(team_winner_guesses)
season_guess.podium_drivers_driver_names_json = json.dumps(podium_driver_guesses)
season_guess.p2_team_id = guesses[1] # type: ignore
season_guess.overtake_driver_id = guesses[2] # type: ignore
season_guess.dnf_driver_id = guesses[3] # type: ignore
season_guess.gained_driver_id = guesses[4] # type: ignore
season_guess.lost_driver_id = guesses[5] # type: ignore
season_guess.team_winners_driver_ids_json = json.dumps(team_winner_guesses)
season_guess.podium_drivers_driver_ids_json = json.dumps(podium_driver_guesses)
db.session.commit()
return redirect(f"/season/Everyone")
def find_or_create_race_result(race_name: str) -> DbRaceResult:
def find_or_create_race_result(race_id: int) -> DbRaceResult:
# There can be a single RaceResult at most, since race_name is the primary key
race_result: DbRaceResult | None = db.session.query(DbRaceResult).filter_by(race_name=race_name).first()
race_result: DbRaceResult | None = db.session.query(DbRaceResult).filter_by(race_id=race_id).first()
if race_result is not None:
return race_result
race_result = DbRaceResult(race_name=race_name,
pxx_driver_names_json=json.dumps(["TEMP"]),
first_dnf_driver_names_json=json.dumps(["TEMP"]),
dnf_driver_names_json=json.dumps(["TEMP"]),
excluded_driver_names_json=json.dumps(["TEMP"]))
race_result = DbRaceResult(race_id=race_id)
race_result.pxx_driver_ids_json = json.dumps(["9999"])
race_result.first_dnf_driver_ids_json = json.dumps(["9999"])
race_result.dnf_driver_ids_json = json.dumps(["9999"])
race_result.excluded_driver_ids_json = json.dumps(["9999"])
db.session.add(race_result)
db.session.commit()
# Double check if database insertion worked and obtain any values set by the database
race_result = db.session.query(DbRaceResult).filter_by(race_name=race_name).first()
race_result = db.session.query(DbRaceResult).filter_by(race_id=race_id).first()
if race_result is None:
raise Exception("Failed adding RaceResult to the database")
return race_result
def update_race_result(race_name: str, pxx_driver_names_list: List[str], first_dnf_driver_names_list: List[str], dnf_driver_names_list: List[str], excluded_driver_names_list: List[str]) -> Response:
if ENABLE_TIMING and not race_has_started(race_name=race_name):
def update_race_result(race_id: int, pxx_driver_ids_list: List[str], first_dnf_driver_ids_list: List[str], dnf_driver_ids_list: List[str], excluded_driver_ids_list: List[str]) -> Response:
if ENABLE_TIMING and not race_has_started(race_id=race_id):
return error_redirect("No race result can be entered, as the race has not begun!")
# Use strings as keys, as these dicts will be serialized to json
pxx_driver_names: Dict[str, str] = {
str(position + 1): driver for position, driver in enumerate(pxx_driver_names_list)
str(position + 1): driver_id for position, driver_id in enumerate(pxx_driver_ids_list)
}
# Not counted drivers have to be at the end
excluded_driver_names: Dict[str, str] = {
str(position + 1): driver for position, driver in enumerate(pxx_driver_names_list)
if driver in excluded_driver_names_list
str(position + 1): driver_id for position, driver_id in enumerate(pxx_driver_ids_list)
if driver_id in excluded_driver_ids_list
}
if len(excluded_driver_names) > 0 and (not "20" in excluded_driver_names or not positions_are_contiguous(list(excluded_driver_names.keys()))):
return error_redirect("Race result was not saved, as excluded drivers must be contiguous and at the end of the field!")
# First DNF drivers have to be contained in DNF drivers
for driver_name in first_dnf_driver_names_list:
if driver_name not in dnf_driver_names_list:
dnf_driver_names_list.append(driver_name)
for driver_id in first_dnf_driver_ids_list:
if driver_id not in dnf_driver_ids_list:
dnf_driver_ids_list.append(driver_id)
# There can't be dnfs but no initial dnfs
if len(dnf_driver_names_list) > 0 and len(first_dnf_driver_names_list) == 0:
if len(dnf_driver_ids_list) > 0 and len(first_dnf_driver_ids_list) == 0:
return error_redirect("Race result was not saved, as there cannot be DNFs without (an) initial DNF(s)!")
race_result: DbRaceResult = find_or_create_race_result(race_name)
race_result.pxx_driver_names_json = json.dumps(pxx_driver_names)
race_result.first_dnf_driver_names_json = json.dumps(first_dnf_driver_names_list)
race_result.dnf_driver_names_json = json.dumps(dnf_driver_names_list)
race_result.excluded_driver_names_json = json.dumps(excluded_driver_names_list)
race_result: DbRaceResult = find_or_create_race_result(race_id)
race_result.pxx_driver_ids_json = json.dumps(pxx_driver_names)
race_result.first_dnf_driver_ids_json = json.dumps(first_dnf_driver_ids_list)
race_result.dnf_driver_ids_json = json.dumps(dnf_driver_ids_list)
race_result.excluded_driver_ids_json = json.dumps(excluded_driver_ids_list)
db.session.commit()
return redirect(f"/result/{quote(race_name)}")
race: DbRace | None = db.session.query(DbRace).filter_by(id=race_id).first()
if race is None:
raise Exception(f"Could not find DbRace with id {race_id}")
return redirect(f"/result/{quote(race.name)}")
def update_user(user_name: str | None, add: bool = False, delete: bool = False) -> Response:
@ -194,7 +203,9 @@ def update_user(user_name: str | None, add: bool = False, delete: bool = False)
disabled_user.enabled = True
else:
user: DbUser = DbUser(name=user_name, enabled=True)
user: DbUser = DbUser(id=None)
user.name = user_name
user.enabled = True
db.session.add(user)
db.session.commit()

14
formula10/database/validation.py
View File

@ -31,19 +31,19 @@ def race_has_started(*, race: Race) -> bool:
return race_has_started(race=race)
@overload
def race_has_started(*, race_name: str) -> bool:
return race_has_started(race_name=race_name)
def race_has_started(*, race_id: int) -> bool:
return race_has_started(race_id=race_id)
def race_has_started(*, race: Race | None = None, race_name: str | None = None) -> bool:
if race is None and race_name is not None:
_race: DbRace | None = db.session.query(DbRace).filter_by(name=race_name).first()
def race_has_started(*, race: Race | None = None, race_id: int | None = None) -> bool:
if race is None and race_id is not None:
_race: DbRace | None = db.session.query(DbRace).filter_by(id=race_id).first()
if _race is None:
raise Exception(f"Couldn't obtain race {race_name} to check date")
raise Exception(f"Couldn't obtain race with id {race_id} to check date")
return datetime.now() > _race.date
if race is not None and race_name is None:
if race is not None and race_id is None:
return datetime.now() > race.date
raise Exception("race_has_started received illegal arguments")