diff --git a/flake.nix b/flake.nix index 47f3899..a64d94d 100644 --- a/flake.nix +++ b/flake.nix @@ -84,6 +84,7 @@ # Web flask flask-sqlalchemy + flask-caching sqlalchemy requests diff --git a/formula10/controller/admin_controller.py b/formula10/controller/admin_controller.py index c0ea31d..7303783 100644 --- a/formula10/controller/admin_controller.py +++ b/formula10/controller/admin_controller.py @@ -5,6 +5,7 @@ from werkzeug import Response from formula10.controller.error_controller import error_redirect from formula10.database.update_queries import update_race_result, update_user +from formula10.domain.cache_invalidator import cache_invalidate_user_updated, cache_invalidate_race_result_updated from formula10.domain.domain_model import Model from formula10.domain.template_model import TemplateModel from formula10 import app @@ -43,6 +44,7 @@ def result_enter_post(race_name: str) -> Response: if fastest_lap is None: return error_redirect("Data was not saved, because fastest lap was not set.") + cache_invalidate_race_result_updated() race_id: int = Model().race_by(race_name=race_name).id return update_race_result(race_id, pxxs, first_dnfs, dnfs, excluded, int(fastest_lap), sprint_pxxs, sprint_dnf_drivers) @@ -55,6 +57,7 @@ def result_fetch_post(race_name: str) -> Response: # @todo Fetch stuff and build the race_result using update_race_result(...) + cache_invalidate_race_result_updated() return redirect("/result") @@ -68,11 +71,13 @@ def user_root() -> str: @app.route("/user-add", methods=["POST"]) def user_add_post() -> Response: + cache_invalidate_user_updated() username: str | None = request.form.get("select-add-user") return update_user(username, add=True) @app.route("/user-delete", methods=["POST"]) def user_delete_post() -> Response: + cache_invalidate_user_updated() username: str | None = request.form.get("select-delete-user") - return update_user(username, delete=True) \ No newline at end of file + return update_user(username, delete=True) diff --git a/formula10/controller/race_controller.py b/formula10/controller/race_controller.py index f4bce01..c16846f 100644 --- a/formula10/controller/race_controller.py +++ b/formula10/controller/race_controller.py @@ -1,8 +1,10 @@ +from typing import List from urllib.parse import unquote from flask import redirect, render_template, request from werkzeug import Response from formula10.database.update_queries import delete_race_guess, update_race_guess +from formula10.domain.cache_invalidator import cache_invalidate_race_guess_updated from formula10.domain.domain_model import Model from formula10.domain.points_model import PointsModel from formula10.domain.template_model import TemplateModel @@ -37,6 +39,7 @@ def race_guess_post(race_name: str, user_name: str) -> Response: pxx: str | None = request.form.get("pxxselect") dnf: str | None = request.form.get("dnfselect") + cache_invalidate_race_guess_updated() race_id: int = Model().race_by(race_name=race_name).id user_id: int = Model().user_by(user_name=user_name).id return update_race_guess(race_id, user_id, @@ -49,6 +52,7 @@ def race_guess_delete_post(race_name: str, user_name: str) -> Response: race_name = unquote(race_name) user_name = unquote(user_name) + cache_invalidate_race_guess_updated() race_id: int = Model().race_by(race_name=race_name).id user_id: int = Model().user_by(user_name=user_name).id return delete_race_guess(race_id, user_id) diff --git a/formula10/controller/season_controller.py b/formula10/controller/season_controller.py index ee86e90..09aa014 100644 --- a/formula10/controller/season_controller.py +++ b/formula10/controller/season_controller.py @@ -5,6 +5,7 @@ from werkzeug import Response from formula10.database.model.db_team import DbTeam from formula10.database.update_queries import update_season_guess +from formula10.domain.cache_invalidator import cache_invalidate_season_guess_updated from formula10.domain.domain_model import Model from formula10.domain.model.team import NONE_TEAM from formula10.domain.points_model import PointsModel @@ -44,5 +45,6 @@ def season_guess_post(user_name: str) -> Response: ] podium_driver_guesses: List[str] = request.form.getlist("podiumdrivers") + cache_invalidate_season_guess_updated() user_id: int = Model().user_by(user_name=user_name).id return update_season_guess(user_id, guesses, team_winner_guesses, podium_driver_guesses) \ No newline at end of file diff --git a/formula10/domain/cache_invalidator.py b/formula10/domain/cache_invalidator.py new file mode 100644 index 0000000..41f9ff6 --- /dev/null +++ b/formula10/domain/cache_invalidator.py @@ -0,0 +1,96 @@ +from typing import List + +from formula10 import cache + + +def cache_invalidate_user_updated() -> None: + caches: List[str] = [ + "domain_all_users", + "domain_all_race_guesses", + "domain_all_season_guesses", + "points_points_per_step", + "points_user_standing", + ] + + memoized_caches: List[str] = [ + "points_by", + "race_guesses_by", + "season_guesses_by", + ] + + for c in caches: + cache.delete(c) + + for c in memoized_caches: + cache.delete_memoized(c) + + +def cache_invalidate_race_result_updated() -> None: + caches: List[str] = [ + "domain_all_race_results", + "points_points_per_step", + "points_team_points_per_step", + "points_dnfs", + "points_driver_points_per_step_cumulative", + "points_wdc_standing_by_position", + "points_wdc_standing_by_driver", + "points_most_dnf_names", + "points_most_gained_names", + "points_most_lost_names", + "points_team_points_per_step_cumulative", + "points_wcc_standing_by_position", + "points_wcc_standing_by_team", + "points_user_standing", + "template_first_race_without_result", + ] + + memoized_caches: List[str] = [ + "driver_points_per_step", + "driver_points_by", + "total_driver_points_by", + "drivers_sorted_by_points", + "total_team_points_by", + "teams_sorted_by_points", + "points_by", + "is_team_winner", + "has_podium", + "picks_with_points_count", + ] + + for c in caches: + cache.delete(c) + + for c in memoized_caches: + cache.delete_memoized(c) + + +def cache_invalidate_race_guess_updated() -> None: + caches: List[str] = [ + "domain_all_race_guesses", + ] + + memoized_caches: List[str] = [ + "race_guesses_by", + ] + + for c in caches: + cache.delete(c) + + for c in memoized_caches: + cache.delete_memoized(c) + + +def cache_invalidate_season_guess_updated() -> None: + caches: List[str] = [ + "domain_all_season_guesses" + ] + + memoized_caches: List[str] = [ + "season_guesses_by" + ] + + for c in caches: + cache.delete(c) + + for c in memoized_caches: + cache.delete_memoized(c) diff --git a/formula10/domain/domain_model.py b/formula10/domain/domain_model.py index 6cbcab2..2849062 100644 --- a/formula10/domain/domain_model.py +++ b/formula10/domain/domain_model.py @@ -18,133 +18,97 @@ from formula10.domain.model.season_guess import SeasonGuess from formula10.domain.model.season_guess_result import SeasonGuessResult from formula10.domain.model.team import NONE_TEAM, Team from formula10.domain.model.user import User -from formula10 import db +from formula10 import db, cache class Model: - _all_users: List[User] | None = None - _all_race_results: List[RaceResult] | None = None - _all_race_guesses: List[RaceGuess] | None = None - _all_season_guesses: List[SeasonGuess] | None = None - _all_season_guess_results: List[SeasonGuessResult] | None = None - _all_races: List[Race] | None = None - _all_drivers: List[Driver] | None = None - _all_active_drivers: List[Driver] | None = None - _all_teams: List[Team] | None = None - - def all_users(self) -> List[User]: + @staticmethod + @cache.cached(timeout=None, key_prefix="domain_all_users") # Clear when adding/deleting users + def all_users() -> List[User]: """ Returns a list of all enabled users. """ - if self._all_users is None: - self._all_users = [ - User.from_db_user(db_user) - for db_user in db.session.query(DbUser).filter_by(enabled=True).all() - ] + db_users = db.session.query(DbUser).filter_by(enabled=True).all() + return [User.from_db_user(db_user) for db_user in db_users] - return self._all_users - - def all_race_results(self) -> List[RaceResult]: + @staticmethod + @cache.cached(timeout=None, key_prefix="domain_all_race_results") # Clear when adding/updating results + def all_race_results() -> List[RaceResult]: """ Returns a list of all race results, in descending order (most recent first). """ - if self._all_race_results is None: - self._all_race_results = [ - RaceResult.from_db_race_result(db_race_result) - for db_race_result in db.session.query(DbRaceResult).join(DbRaceResult.race).order_by(desc("number")).all() - ] + db_race_results = db.session.query(DbRaceResult).join(DbRaceResult.race).order_by(desc("number")).all() + return [RaceResult.from_db_race_result(db_race_result) for db_race_result in db_race_results] - return self._all_race_results - - def all_race_guesses(self) -> List[RaceGuess]: + @staticmethod + @cache.cached(timeout=None, key_prefix="domain_all_race_guesses") # Clear when adding/updating race guesses or users + def all_race_guesses() -> List[RaceGuess]: """ Returns a list of all race guesses (of enabled users). """ - if self._all_race_guesses is None: - self._all_race_guesses = [ - RaceGuess.from_db_race_guess(db_race_guess) - for db_race_guess in db.session.query(DbRaceGuess).join(DbRaceGuess.user).filter_by(enabled=True).all() # Ignore disabled users - ] + db_race_guesses = db.session.query(DbRaceGuess).join(DbRaceGuess.user).filter_by(enabled=True).all() + return [RaceGuess.from_db_race_guess(db_race_guess) for db_race_guess in db_race_guesses] - return self._all_race_guesses - - def all_season_guesses(self) -> List[SeasonGuess]: + @staticmethod + @cache.cached(timeout=None, key_prefix="domain_all_season_guesses") # Clear when adding/updating season guesses or users + def all_season_guesses() -> List[SeasonGuess]: """ Returns a list of all season guesses (of enabled users). """ - if self._all_season_guesses is None: - self._all_season_guesses = [ - SeasonGuess.from_db_season_guess(db_season_guess) - for db_season_guess in db.session.query(DbSeasonGuess).join(DbSeasonGuess.user).filter_by(enabled=True).all() # Ignore disabled users - ] + db_season_guesses = db.session.query(DbSeasonGuess).join(DbSeasonGuess.user).filter_by(enabled=True).all() + return [SeasonGuess.from_db_season_guess(db_season_guess) for db_season_guess in db_season_guesses] - return self._all_season_guesses + @staticmethod + @cache.cached(timeout=None, key_prefix="domain_all_season_guess_results") # No cleanup, bc entered manually + def all_season_guess_results() -> List[SeasonGuessResult]: + """ + Returns a list of all season guess results (of enabled users). + """ + db_season_guess_results = db.session.query(DbSeasonGuessResult).join(DbSeasonGuessResult.user).filter_by(enabled=True).all() + return [SeasonGuessResult.from_db_season_guess_result(db_season_guess_result) for db_season_guess_result in db_season_guess_results] - def all_season_guess_results(self) -> List[SeasonGuessResult]: - if self._all_season_guess_results is None: - self._all_season_guess_results = [ - SeasonGuessResult.from_db_season_guess_result(db_season_guess_result) - for db_season_guess_result in db.session.query(DbSeasonGuessResult).join(DbSeasonGuessResult.user).filter_by(enabled=True).all() # Ignore disabled users - ] - - return self._all_season_guess_results - - def all_races(self) -> List[Race]: + @staticmethod + @cache.cached(timeout=None, key_prefix="domain_all_races") # No cleanup, bc entered manually + def all_races() -> List[Race]: """ Returns a list of all races, in descending order (last race first). """ - if self._all_races is None: - self._all_races = [ - Race.from_db_race(db_race) - for db_race in db.session.query(DbRace).order_by(desc("number")).all() - ] + db_races = db.session.query(DbRace).order_by(desc("number")).all() + return [Race.from_db_race(db_race) for db_race in db_races] - return self._all_races - - def all_drivers(self, *, include_none: bool, include_inactive: bool) -> List[Driver]: + @staticmethod + @cache.memoize(timeout=None) # No cleanup, bc entered manually + def all_drivers(*, include_none: bool, include_inactive: bool) -> List[Driver]: """ Returns a list of all active drivers. """ - if include_inactive: - if self._all_drivers is None: - self._all_drivers = [ - Driver.from_db_driver(db_driver) - for db_driver in db.session.query(DbDriver).all() - ] + db_drivers = db.session.query(DbDriver).all() + drivers = [Driver.from_db_driver(db_driver) for db_driver in db_drivers] - if include_none: - return self._all_drivers - else: - predicate: Callable[[Driver], bool] = lambda driver: driver != NONE_DRIVER - return find_multiple_strict(predicate, self._all_drivers) - else: - if self._all_active_drivers is None: - self._all_active_drivers = [ - Driver.from_db_driver(db_driver) - for db_driver in db.session.query(DbDriver).filter_by(active=True).all() - ] + if not include_inactive: + predicate: Callable[[Driver], bool] = lambda driver: driver.active + drivers = find_multiple_strict(predicate, drivers) - if include_none: - return self._all_active_drivers - else: - predicate: Callable[[Driver], bool] = lambda driver: driver != NONE_DRIVER - return find_multiple_strict(predicate, self._all_active_drivers) + if not include_none: + predicate: Callable[[Driver], bool] = lambda driver: driver != NONE_DRIVER + drivers = find_multiple_strict(predicate, drivers) - def all_teams(self, *, include_none: bool) -> List[Team]: + return drivers + + @staticmethod + @cache.memoize(timeout=None) # No cleanup, bc entered manually + def all_teams(*, include_none: bool) -> List[Team]: """ Returns a list of all teams. """ - if self._all_teams is None: - self._all_teams = [ - Team.from_db_team(db_team) - for db_team in db.session.query(DbTeam).all() - ] + db_teams = db.session.query(DbTeam).all() + teams = [Team.from_db_team(db_team) for db_team in db_teams] - if include_none: - return self._all_teams - else: + if not include_none: predicate: Callable[[Team], bool] = lambda team: team != NONE_TEAM - return find_multiple_strict(predicate, self._all_teams) + return find_multiple_strict(predicate, teams) + + return teams # # User queries @@ -217,6 +181,7 @@ class Model: """ return self.race_guesses_by() + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race guesses or users def race_guesses_by(self, *, user_name: str | None = None, race_name: str | None = None) -> RaceGuess | List[RaceGuess] | Dict[str, Dict[str, RaceGuess]] | None: # List of all guesses by a single user if user_name is not None and race_name is None: @@ -266,6 +231,7 @@ class Model: """ return self.season_guesses_by() + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating season guesses or users def season_guesses_by(self, *, user_name: str | None = None) -> SeasonGuess | Dict[str, SeasonGuess] | None: if user_name is not None: predicate: Callable[[SeasonGuess], bool] = lambda guess: guess.user.name == user_name @@ -320,6 +286,7 @@ class Model: """ return self.drivers_by(include_inactive=include_inactive) + @cache.memoize(timeout=None, args_to_ignore=["self"]) # No Cleanup, data added manually def drivers_by(self, *, team_name: str | None = None, include_inactive: bool) -> List[Driver] | Dict[str, List[Driver]]: if team_name is not None: predicate: Callable[[Driver], bool] = lambda driver: driver.team.name == team_name diff --git a/formula10/domain/model/driver.py b/formula10/domain/model/driver.py index f86426b..5361e89 100644 --- a/formula10/domain/model/driver.py +++ b/formula10/domain/model/driver.py @@ -34,6 +34,11 @@ class Driver: def __hash__(self) -> int: return hash(self.id) + # This is important to memoize functions getting a Driver as input. + # The repr() will be appended to the cache key. + def __repr__(self) -> str: + return f"Driver(id={self.id}, name={self.name})" + id: int name: str abbr: str @@ -52,4 +57,4 @@ NONE_DRIVER.name = "None" NONE_DRIVER.abbr = "None" NONE_DRIVER.country = "NO" NONE_DRIVER.team = NONE_TEAM -NONE_DRIVER.active = False \ No newline at end of file +NONE_DRIVER.active = True \ No newline at end of file diff --git a/formula10/domain/points_model.py b/formula10/domain/points_model.py index cf5b1c2..aefa590 100644 --- a/formula10/domain/points_model.py +++ b/formula10/domain/points_model.py @@ -2,6 +2,7 @@ import json from typing import Any, Callable, Dict, List, overload import numpy as np +from formula10 import cache from formula10.domain.domain_model import Model from formula10.domain.model.driver import NONE_DRIVER, Driver from formula10.domain.model.race_guess import RaceGuess @@ -119,118 +120,93 @@ class PointsModel(Model): This class bundles all data + functionality required to do points calculations. """ - _points_per_step: Dict[str, List[int]] | None = None - _driver_points_per_step: Dict[str, List[int]] | None = None - _active_driver_points_per_step: Dict[str, List[int]] | None = None - _team_points_per_step: Dict[str, List[int]] | None = None - _dnfs: Dict[str, int] | None = None - def __init__(self): Model.__init__(self) + @cache.cached(timeout=None, key_prefix="points_points_per_step") # Clear when adding/updating race results or users def points_per_step(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing points per race for each user. """ - if self._points_per_step is None: - self._points_per_step = dict() - for user in self.all_users(): - self._points_per_step[user.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers + points_per_step = dict() + for user in self.all_users(): + points_per_step[user.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers - for race_guess in self.all_race_guesses(): - user_name: str = race_guess.user.name - race_number: int = race_guess.race.number - race_result: RaceResult | None = self.race_result_by(race_name=race_guess.race.name) + for race_guess in self.all_race_guesses(): + user_name: str = race_guess.user.name + race_number: int = race_guess.race.number + race_result: RaceResult | None = self.race_result_by(race_name=race_guess.race.name) - if race_result is None: - continue + if race_result is None: + continue - self._points_per_step[user_name][race_number] = standing_points(race_guess, race_result) + dnf_points(race_guess, race_result) + points_per_step[user_name][race_number] = standing_points(race_guess, race_result) + dnf_points(race_guess, race_result) - return self._points_per_step + return points_per_step + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Clear when adding/updating race results def driver_points_per_step(self, *, include_inactive: bool) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing points per race for each driver. """ - if include_inactive: - if self._driver_points_per_step is None: - self._driver_points_per_step = dict() - for driver in self.all_drivers(include_none=False, include_inactive=True): - self._driver_points_per_step[driver.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers + driver_points_per_step = dict() + for driver in self.all_drivers(include_none=False, include_inactive=include_inactive): + driver_points_per_step[driver.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers - for race_result in self.all_race_results(): - race_number: int = race_result.race.number + for race_result in self.all_race_results(): + race_number: int = race_result.race.number - for position, driver in race_result.standing.items(): - self._driver_points_per_step[driver.name][race_number] = DRIVER_RACE_POINTS[int(position)] if int(position) in DRIVER_RACE_POINTS else 0 - self._driver_points_per_step[driver.name][race_number] += DRIVER_FASTEST_LAP_POINTS if race_result.fastest_lap_driver == driver else 0 + for position, driver in race_result.standing.items(): + driver_points_per_step[driver.name][race_number] = DRIVER_RACE_POINTS[int(position)] if int(position) in DRIVER_RACE_POINTS else 0 + driver_points_per_step[driver.name][race_number] += DRIVER_FASTEST_LAP_POINTS if race_result.fastest_lap_driver == driver else 0 - for position, driver in race_result.sprint_standing.items(): - driver_name: str = driver.name + for position, driver in race_result.sprint_standing.items(): + driver_name: str = driver.name - self._driver_points_per_step[driver_name][race_number] += DRIVER_SPRINT_POINTS[int(position)] if int(position) in DRIVER_SPRINT_POINTS else 0 + driver_points_per_step[driver_name][race_number] += DRIVER_SPRINT_POINTS[int(position)] if int(position) in DRIVER_SPRINT_POINTS else 0 - return self._driver_points_per_step - else: - if self._active_driver_points_per_step is None: - self._active_driver_points_per_step = dict() - for driver in self.all_drivers(include_none=False, include_inactive=False): - self._active_driver_points_per_step[driver.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers - - for race_result in self.all_race_results(): - race_number: int = race_result.race.number - - for position, driver in race_result.standing.items(): - self._active_driver_points_per_step[driver.name][race_number] = DRIVER_RACE_POINTS[int(position)] if int(position) in DRIVER_RACE_POINTS else 0 - self._active_driver_points_per_step[driver.name][race_number] += DRIVER_FASTEST_LAP_POINTS if race_result.fastest_lap_driver == driver else 0 - - for position, driver in race_result.sprint_standing.items(): - driver_name: str = driver.name - - self._active_driver_points_per_step[driver_name][race_number] += DRIVER_SPRINT_POINTS[int(position)] if int(position) in DRIVER_SPRINT_POINTS else 0 - - return self._active_driver_points_per_step + return driver_points_per_step + @cache.cached(timeout=None, key_prefix="points_team_points_per_step") def team_points_per_step(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing points per race for each team. """ - if self._team_points_per_step is None: - self._team_points_per_step = dict() - for team in self.all_teams(include_none=False): - self._team_points_per_step[team.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers + team_points_per_step = dict() + for team in self.all_teams(include_none=False): + team_points_per_step[team.name] = [0] * (len(self.all_races()) + 1) # Start at index 1, like the race numbers - for race_result in self.all_race_results(): - for driver in race_result.standing.values(): - team_name: str = driver.team.name - race_number: int = race_result.race.number + for race_result in self.all_race_results(): + for driver in race_result.standing.values(): + team_name: str = driver.team.name + race_number: int = race_result.race.number - self._team_points_per_step[team_name][race_number] += self.driver_points_per_step(include_inactive=True)[driver.name][race_number] + team_points_per_step[team_name][race_number] += self.driver_points_per_step(include_inactive=True)[driver.name][race_number] - return self._team_points_per_step + return team_points_per_step + @cache.cached(timeout=None, key_prefix="points_dnfs") def dnfs(self) -> Dict[str, int]: - if self._dnfs is None: - self._dnfs = dict() + dnfs = dict() - for driver in self.all_drivers(include_none=False, include_inactive=True): - self._dnfs[driver.name] = 0 + for driver in self.all_drivers(include_none=False, include_inactive=True): + dnfs[driver.name] = 0 - for race_result in self.all_race_results(): - for driver in race_result.all_dnfs: - self._dnfs[driver.name] += 1 + for race_result in self.all_race_results(): + for driver in race_result.all_dnfs: + dnfs[driver.name] += 1 - for driver in race_result.sprint_dnfs: - self._dnfs[driver.name] += 1 + for driver in race_result.sprint_dnfs: + dnfs[driver.name] += 1 - return self._dnfs + return dnfs # # Driver stats # + @cache.cached(timeout=None, key_prefix="points_driver_points_per_step_cumulative") # Cleanup when adding/updating race results def driver_points_per_step_cumulative(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing cumulative points per race for each driver. @@ -262,6 +238,7 @@ class PointsModel(Model): """ return self.driver_points_by(driver_name=driver_name, race_name=race_name, include_inactive=include_inactive) + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def driver_points_by(self, *, driver_name: str | None = None, race_name: str | None = None, include_inactive: bool) -> List[int] | Dict[str, int] | int: if driver_name is not None and race_name is None: return self.driver_points_per_step(include_inactive=include_inactive)[driver_name] @@ -282,13 +259,16 @@ class PointsModel(Model): raise Exception("driver_points_by received an illegal combination of arguments") + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def total_driver_points_by(self, driver_name: str) -> int: return sum(self.driver_points_by(driver_name=driver_name, include_inactive=True)) + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def drivers_sorted_by_points(self, *, include_inactive: bool) -> List[Driver]: comparator: Callable[[Driver], int] = lambda driver: self.total_driver_points_by(driver.name) return sorted(self.all_drivers(include_none=False, include_inactive=include_inactive), key=comparator, reverse=True) + @cache.cached(timeout=None, key_prefix="points_wdc_standing_by_position") # Cleanup when adding/updating race results def wdc_standing_by_position(self) -> Dict[int, List[str]]: standing: Dict[int, List[str]] = dict() @@ -308,6 +288,7 @@ class PointsModel(Model): return standing + @cache.cached(timeout=None, key_prefix="points_wdc_standing_by_driver") # Cleanup when adding/updating race results def wdc_standing_by_driver(self) -> Dict[str, int]: standing: Dict[str, int] = dict() @@ -330,6 +311,7 @@ class PointsModel(Model): return WDC_STANDING_2023[driver_name] - self.wdc_standing_by_driver()[driver_name] + @cache.cached(timeout=None, key_prefix="points_most_dnf_names") # Cleanup when adding/updating race results def most_dnf_names(self) -> List[str]: dnf_names: List[str] = list() most_dnfs: int = 0 @@ -344,6 +326,7 @@ class PointsModel(Model): return dnf_names + @cache.cached(timeout=None, key_prefix="points_most_gained_names") # Cleanup when adding/updating race results def most_gained_names(self) -> List[str]: most_gained_names: List[str] = list() most_gained: int = 0 @@ -362,6 +345,7 @@ class PointsModel(Model): return most_gained_names + @cache.cached(timeout=None, key_prefix="points_most_lost_names") # Cleanup when adding/updating race results def most_lost_names(self) -> List[str]: most_lost_names: List[str] = list() most_lost: int = 100 @@ -384,6 +368,7 @@ class PointsModel(Model): # Team points # + @cache.cached(timeout=None, key_prefix="points_team_points_per_step_cumulative") # Cleanup when adding/updating race results def team_points_per_step_cumulative(self) -> Dict[str, List[int]]: """ Returns a dictionary of lists, containing cumulative points per race for each team. @@ -394,14 +379,17 @@ class PointsModel(Model): return points_per_step_cumulative + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def total_team_points_by(self, team_name: str) -> int: teammates: List[Driver] = self.drivers_by(team_name=team_name, include_inactive=True) return sum(sum(self.driver_points_by(driver_name=teammate.name, include_inactive=True)) for teammate in teammates) + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def teams_sorted_by_points(self) -> List[Team]: comparator: Callable[[Team], int] = lambda team: self.total_team_points_by(team.name) return sorted(self.all_teams(include_none=False), key=comparator, reverse=True) + @cache.cached(timeout=None, key_prefix="points_wcc_standing_by_position") # Cleanup when adding/updating race results def wcc_standing_by_position(self) -> Dict[int, List[str]]: standing: Dict[int, List[str]] = dict() @@ -421,6 +409,7 @@ class PointsModel(Model): return standing + @cache.cached(timeout=None, key_prefix="points_wcc_standing_by_team") # Cleanup when adding/updating race results def wcc_standing_by_team(self) -> Dict[str, int]: standing: Dict[str, int] = dict() @@ -475,6 +464,7 @@ class PointsModel(Model): """ return self.points_by(user_name=user_name, race_name=race_name) + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results or users def points_by(self, *, user_name: str | None = None, race_name: str | None = None) -> List[int] | Dict[str, int] | int: if user_name is not None and race_name is None: return self.points_per_step()[user_name] @@ -508,6 +498,7 @@ class PointsModel(Model): comparator: Callable[[User], int] = lambda user: self.total_points_by(user.name) return sorted(self.all_users(), key=comparator, reverse=True) + @cache.cached(timeout=None, key_prefix="points_user_standing") # Cleanup when adding/updating race results or users def user_standing(self) -> Dict[str, int]: standing: Dict[str, int] = dict() @@ -527,6 +518,7 @@ class PointsModel(Model): # Treat standing + dnf picks separately return len(self.race_guesses_by(user_name=user_name)) * 2 + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def picks_with_points_count(self, user_name: str) -> int: count: int = 0 @@ -594,14 +586,14 @@ class PointsModel(Model): return season_guess.most_wdc_lost.name in self.most_lost_names() + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def is_team_winner(self, driver: Driver) -> bool: teammates: List[Driver] = self.drivers_by(team_name=driver.team.name, include_inactive=True) teammate: Driver = teammates[0] if teammates[1] == driver else teammates[1] - print(f"{driver.name} standing: {self.wdc_standing_by_driver()[driver.name]}, {teammate.name} standing: {self.wdc_standing_by_driver()[teammate.name]}") - return self.wdc_standing_by_driver()[driver.name] <= self.wdc_standing_by_driver()[teammate.name] + @cache.memoize(timeout=None, args_to_ignore=["self"]) # Cleanup when adding/updating race results def has_podium(self, driver: Driver) -> bool: for race_result in self.all_race_results(): position: int | None = race_result.driver_standing_position(driver) diff --git a/formula10/domain/template_model.py b/formula10/domain/template_model.py index c69ecdd..1804118 100644 --- a/formula10/domain/template_model.py +++ b/formula10/domain/template_model.py @@ -1,5 +1,5 @@ from typing import List, Callable -from formula10 import ENABLE_TIMING +from formula10 import ENABLE_TIMING, cache from formula10.domain.domain_model import Model from formula10.domain.model.driver import Driver @@ -54,6 +54,7 @@ class TemplateModel(Model): return self.all_users() + @cache.cached(timeout=None, key_prefix="template_first_race_without_result") # Cleanup when adding/updating race results def first_race_without_result(self) -> Race | None: """ Returns the first race-object with no associated race result. diff --git a/requirements.txt b/requirements.txt index ae61a6b..6fe4217 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,7 @@ numpy flask flask-sqlalchemy +flask-caching sqlalchemy requests werkzeug