import math from fastapi import APIRouter, Depends, HTTPException, Query, Response from typing import List, Optional, Literal import logging import pandas as pd from pydantic import BaseModel, validator from ..db_engine import ( db, StratPlay, StratGame, Team, Player, model_to_dict, chunked, fn, SQL, Case, complex_data_to_csv, Decision, ) from ..db_helpers import upsert_strat_plays from ..dependencies import oauth2_scheme, valid_token, LOG_DATA logging.basicConfig( filename=LOG_DATA["filename"], format=LOG_DATA["format"], level=LOG_DATA["log_level"], ) router = APIRouter(prefix="/api/v2/plays", tags=["plays"]) POS_LIST = Literal[ "C", "1B", "2B", "3B", "SS", "LF", "CF", "RF", "P", "DH", "PH", "PR", "GHOST" ] class PlayModel(BaseModel): game_id: int play_num: int batter_id: int = None batter_team_id: int = None pitcher_id: int pitcher_team_id: int = None on_base_code: str inning_half: Literal["top", "bot", "Top", "Bot"] inning_num: int batting_order: int starting_outs: int away_score: int home_score: int batter_pos: POS_LIST = None on_first_id: int = None on_first_final: int = None on_second_id: int = None on_second_final: int = None on_third_id: int = None on_third_final: int = None batter_final: int = None pa: int = 0 ab: int = 0 e_run: int = 0 run: int = 0 hit: int = 0 rbi: int = 0 double: int = 0 triple: int = 0 homerun: int = 0 bb: int = 0 so: int = 0 hbp: int = 0 sac: int = 0 ibb: int = 0 gidp: int = 0 bphr: int = 0 bpfo: int = 0 bp1b: int = 0 bplo: int = 0 sb: int = 0 cs: int = 0 outs: int = 0 wpa: float = 0.0 re24: float = 0.0 catcher_id: int = None catcher_team_id: int = None defender_id: int = None defender_team_id: int = None runner_id: int = None runner_team_id: int = None check_pos: POS_LIST = None error: int = 0 wild_pitch: int = 0 passed_ball: int = 0 pick_off: int = 0 balk: int = 0 is_go_ahead: bool = False is_tied: bool = False is_new_inning: bool = False @validator("on_first_final") def no_final_if_no_runner_one(cls, v, values): if values["on_first_id"] is None: return None return v @validator("on_second_final") def no_final_if_no_runner_two(cls, v, values): if values["on_second_id"] is None: return None return v @validator("on_third_final") def no_final_if_no_runner_three(cls, v, values): if values["on_third_id"] is None: return None return v @validator("batter_final") def no_final_if_no_batter(cls, v, values): if values["batter_id"] is None: return None return v class PlayList(BaseModel): plays: List[PlayModel] @router.get("") async def get_plays( game_id: list = Query(default=None), batter_id: list = Query(default=None), season: list = Query(default=None), week: list = Query(default=None), has_defender: Optional[bool] = None, has_catcher: Optional[bool] = None, has_defender_or_catcher: Optional[bool] = None, is_scoring_play: Optional[bool] = None, pitcher_id: list = Query(default=None), obc: list = Query(default=None), inning: list = Query(default=None), batting_order: list = Query(default=None), starting_outs: list = Query(default=None), batter_pos: list = Query(default=None), catcher_id: list = Query(default=None), defender_id: list = Query(default=None), runner_id: list = Query(default=None), offense_team_id: list = Query(default=None), defense_team_id: list = Query(default=None), hit: Optional[int] = None, double: Optional[int] = None, triple: Optional[int] = None, homerun: Optional[int] = None, play_num: list = Query(default=None), game_type: list = Query(default=None), sb: Optional[int] = None, cs: Optional[int] = None, csv: Optional[bool] = False, run: Optional[int] = None, e_run: Optional[int] = None, rbi: list = Query(default=None), outs: list = Query(default=None), wild_pitch: Optional[int] = None, is_final_out: Optional[bool] = None, is_go_ahead: Optional[bool] = None, is_tied: Optional[bool] = None, is_new_inning: Optional[bool] = None, min_wpa: Optional[float] = None, max_wpa: Optional[float] = None, sort: Optional[str] = None, short_output: Optional[bool] = False, limit: Optional[int] = 200, page_num: Optional[int] = 1, ): all_plays = StratPlay.select().order_by(StratPlay.id) if season is not None: s_games = StratGame.select().where(StratGame.season << season) all_plays = all_plays.where(StratPlay.game << s_games) if week is not None: w_games = StratGame.select().where(StratGame.week << week) all_plays = all_plays.where(StratPlay.game << w_games) if has_defender is not None: all_plays = all_plays.where(StratPlay.defender.is_null(False)) if has_catcher is not None: all_plays = all_plays.where(StratPlay.catcher.is_null(False)) if has_defender_or_catcher is not None: all_plays = all_plays.where( (StratPlay.catcher.is_null(False)) | (StratPlay.defender.is_null(False)) ) if game_id is not None: all_plays = all_plays.where(StratPlay.game_id << game_id) if batter_id is not None: all_plays = all_plays.where(StratPlay.batter_id << batter_id) if pitcher_id is not None: all_plays = all_plays.where(StratPlay.pitcher_id << pitcher_id) if obc is not None: all_plays = all_plays.where(StratPlay.on_base_code << obc) if inning is not None: all_plays = all_plays.where(StratPlay.inning_num << inning) if batting_order is not None: all_plays = all_plays.where(StratPlay.batting_order << batting_order) if starting_outs is not None: all_plays = all_plays.where(StratPlay.starting_outs << starting_outs) if batter_pos is not None: all_plays = all_plays.where(StratPlay.batter_pos << batter_pos) if catcher_id is not None: all_plays = all_plays.where(StratPlay.catcher_id << catcher_id) if defender_id is not None: all_plays = all_plays.where(StratPlay.defender_id << defender_id) if runner_id is not None: all_plays = all_plays.where(StratPlay.runner_id << runner_id) if offense_team_id is not None: all_teams = Team.select().where(Team.id << offense_team_id) all_plays = all_plays.where( (StratPlay.batter_team << all_teams) | (StratPlay.runner_team << all_teams) ) if defense_team_id is not None: all_teams = Team.select().where(Team.id << defense_team_id) all_plays = all_plays.where( (StratPlay.catcher_team << all_teams) | (StratPlay.defender_team << all_teams) ) if hit is not None: all_plays = all_plays.where(StratPlay.hit == hit) if double is not None: all_plays = all_plays.where(StratPlay.double == double) if triple is not None: all_plays = all_plays.where(StratPlay.triple == triple) if homerun is not None: all_plays = all_plays.where(StratPlay.homerun == homerun) if sb is not None: all_plays = all_plays.where(StratPlay.sb == sb) if cs is not None: all_plays = all_plays.where(StratPlay.cs == cs) if wild_pitch is not None: all_plays = all_plays.where(StratPlay.wild_pitch == wild_pitch) if run is not None: all_plays = all_plays.where(StratPlay.run == run) if e_run is not None: all_plays = all_plays.where(StratPlay.e_run == e_run) if rbi is not None: all_plays = all_plays.where(StratPlay.rbi << rbi) if outs is not None: all_plays = all_plays.where(StratPlay.outs << outs) if is_final_out is not None: all_plays = all_plays.where(StratPlay.starting_outs + StratPlay.outs == 3) if is_go_ahead is not None: all_plays = all_plays.where(StratPlay.is_go_ahead == is_go_ahead) if is_tied is not None: all_plays = all_plays.where(StratPlay.is_tied == is_tied) if is_new_inning is not None: all_plays = all_plays.where(StratPlay.is_new_inning == is_new_inning) if is_scoring_play is not None: all_plays = all_plays.where( (StratPlay.on_first_final == 4) | (StratPlay.on_second_final == 4) | (StratPlay.on_third_final == 4) | (StratPlay.batter_final == 4) ) if min_wpa is not None: all_plays = all_plays.where(StratPlay.wpa >= min_wpa) if max_wpa is not None: all_plays = all_plays.where(StratPlay.wpa <= max_wpa) if play_num is not None: all_plays = all_plays.where(StratPlay.play_num << play_num) if game_type is not None: all_types = [x.lower() for x in game_type] all_games = StratGame.select().where(fn.Lower(StratGame.game_type) << all_types) all_plays = all_plays.where(StratPlay.game << all_games) if limit > 5000: limit = 5000 elif limit < 1: limit = 1 if page_num < 1: page_num = 1 if sort == "wpa-desc": all_plays = all_plays.order_by(-fn.ABS(StratPlay.wpa)) elif sort == "wpa-asc": all_plays = all_plays.order_by(fn.ABS(StratPlay.wpa)) elif sort == "re24-desc": all_plays = all_plays.order_by(-fn.ABS(StratPlay.re24)) elif sort == "re24-asc": all_plays = all_plays.order_by(fn.ABS(StratPlay.re24)) elif sort == "newest": all_plays = all_plays.order_by( StratPlay.game_id.desc(), StratPlay.play_num.desc() ) elif sort == "oldest": all_plays = all_plays.order_by(StratPlay.game_id, StratPlay.play_num) all_plays = all_plays.paginate(page_num, limit) if csv: return_vals = [model_to_dict(x) for x in all_plays] for x in return_vals: x["game_id"] = x["game"]["id"] x["game_type"] = x["game"]["game_type"] x["batter_id"] = x["batter"]["player_id"] x["batter_name"] = x["batter"]["p_name"] x["batter_cardset"] = x["batter"]["cardset"]["name"] x["batter_team_id"] = x["batter_team"]["id"] x["batter_team_abbrev"] = x["batter_team"]["abbrev"] x["pitcher_id"] = x["pitcher"]["player_id"] x["pitcher_name"] = x["pitcher"]["p_name"] x["pitcher_cardset"] = x["pitcher"]["cardset"]["name"] x["pitcher_team_id"] = x["pitcher_team"]["id"] x["pitcher_team_abbrev"] = x["pitcher_team"]["abbrev"] if x["catcher"] is not None: x["catcher_id"] = x["catcher"]["player_id"] x["catcher_name"] = x["catcher"]["p_name"] x["catcher_cardset"] = x["catcher"]["cardset"]["name"] x["catcher_team_id"] = x["catcher_team"]["id"] x["catcher_team_abbrev"] = x["catcher_team"]["abbrev"] else: x["catcher_id"] = None x["catcher_name"] = None x["catcher_cardset"] = None x["catcher_team_id"] = None x["catcher_team_abbrev"] = None if x["defender"] is not None: x["defender_id"] = x["defender"]["player_id"] x["defender_name"] = x["defender"]["p_name"] x["defender_cardset"] = x["defender"]["cardset"]["name"] x["defender_team_id"] = x["defender_team"]["id"] x["defender_team_abbrev"] = x["defender_team"]["abbrev"] else: x["defender_id"] = None x["defender_name"] = None x["defender_cardset"] = None x["defender_team_id"] = None x["defender_team_abbrev"] = None if x["runner"] is not None: x["runner_id"] = x["runner"]["player_id"] x["runner_name"] = x["runner"]["p_name"] x["runner_cardset"] = x["runner"]["cardset"]["name"] x["runner_team_id"] = x["runner_team"]["id"] x["runner_team_abbrev"] = x["runner_team"]["abbrev"] else: x["runner_id"] = None x["runner_name"] = None x["runner_cardset"] = None x["runner_team_id"] = None x["runner_team_abbrev"] = None del ( x["game"], x["batter"], x["batter_team"], x["pitcher"], x["pitcher_team"], x["catcher"], x["catcher_team"], x["defender"], x["defender_team"], x["runner"], x["runner_team"], ) db.close() return Response( content=pd.DataFrame(return_vals).to_csv(index=False), media_type="text/csv" ) return_plays = { "count": all_plays.count(), "plays": [model_to_dict(x, recurse=not short_output) for x in all_plays], } db.close() return return_plays @router.get("/batting") async def get_batting_totals( season: list = Query(default=None), week: list = Query(default=None), position: list = Query(default=None), player_id: list = Query(default=None), min_wpa: Optional[float] = -999, max_wpa: Optional[float] = 999, group_by: Literal[ "team", "player", "playerteam", "playergame", "teamgame", "league", "gmtype", "playergtype", "playerteamgtype", ] = "player", is_gauntlet: Optional[bool] = None, min_pa: Optional[int] = 1, team_id: list = Query(default=None), inning: list = Query(default=None), obc: list = Query(default=None), risp: Optional[bool] = None, game_type: list = Query(default=None), page_num: Optional[int] = 1, sort: Optional[str] = "pa-desc", limit: Optional[int] = 500, short_output: Optional[bool] = False, csv: Optional[bool] = False, ): season_games = StratGame.select() if season is not None: season_games = season_games.where(StratGame.season << season) if week is not None: season_games = season_games.where(StratGame.week << week) # Build SELECT fields conditionally based on group_by to satisfy PostgreSQL's # strict GROUP BY requirement (all non-aggregated SELECT fields must be in GROUP BY) base_select_fields = [ fn.SUM(StratPlay.pa).alias("sum_pa"), fn.SUM(StratPlay.ab).alias("sum_ab"), fn.SUM(StratPlay.run).alias("sum_run"), fn.SUM(StratPlay.hit).alias("sum_hit"), fn.SUM(StratPlay.rbi).alias("sum_rbi"), fn.SUM(StratPlay.double).alias("sum_double"), fn.SUM(StratPlay.triple).alias("sum_triple"), fn.SUM(StratPlay.homerun).alias("sum_hr"), fn.SUM(StratPlay.bb).alias("sum_bb"), fn.SUM(StratPlay.so).alias("sum_so"), fn.SUM(StratPlay.hbp).alias("sum_hbp"), fn.SUM(StratPlay.sac).alias("sum_sac"), fn.SUM(StratPlay.ibb).alias("sum_ibb"), fn.SUM(StratPlay.gidp).alias("sum_gidp"), fn.SUM(StratPlay.sb).alias("sum_sb"), fn.SUM(StratPlay.cs).alias("sum_cs"), fn.SUM(StratPlay.bphr).alias("sum_bphr"), fn.SUM(StratPlay.bpfo).alias("sum_bpfo"), fn.SUM(StratPlay.bp1b).alias("sum_bp1b"), fn.SUM(StratPlay.bplo).alias("sum_bplo"), fn.SUM(StratPlay.wpa).alias("sum_wpa"), fn.SUM(StratPlay.re24).alias("sum_re24"), fn.COUNT(StratPlay.on_first_final) .filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4) ) .alias("count_lo1"), fn.COUNT(StratPlay.on_second_final) .filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4) ) .alias("count_lo2"), fn.COUNT(StratPlay.on_third_final) .filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4) ) .alias("count_lo3"), fn.COUNT(StratPlay.on_first) .filter(StratPlay.on_first.is_null(False)) .alias("count_runner1"), fn.COUNT(StratPlay.on_second) .filter(StratPlay.on_second.is_null(False)) .alias("count_runner2"), fn.COUNT(StratPlay.on_third) .filter(StratPlay.on_third.is_null(False)) .alias("count_runner3"), fn.COUNT(StratPlay.on_first_final) .filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3) ) .alias("count_lo1_3out"), fn.COUNT(StratPlay.on_second_final) .filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3) ) .alias("count_lo2_3out"), fn.COUNT(StratPlay.on_third_final) .filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3) ) .alias("count_lo3_3out"), ] # Add non-aggregated fields based on grouping type if group_by in [ "player", "playerteam", "playergame", "playergtype", "playerteamgtype", ]: base_select_fields.insert(0, StratPlay.batter) if group_by in ["team", "playerteam", "teamgame", "playerteamgtype"]: base_select_fields.append(StratPlay.batter_team) if group_by in ["playergame", "teamgame"]: base_select_fields.append(StratPlay.game) bat_plays = ( StratPlay.select(*base_select_fields) .where((StratPlay.game << season_games) & (StratPlay.batter.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_pa) ) # Build run_plays SELECT fields conditionally run_select_fields = [ fn.SUM(StratPlay.sb).alias("sum_sb"), fn.SUM(StratPlay.cs).alias("sum_cs"), fn.SUM(StratPlay.pick_off).alias("sum_pick"), fn.SUM(StratPlay.wpa).alias("sum_wpa"), fn.SUM(StratPlay.re24).alias("sum_re24"), ] if group_by in [ "player", "playerteam", "playergame", "playergtype", "playerteamgtype", ]: run_select_fields.insert(0, StratPlay.runner) if group_by in ["team", "playerteam", "teamgame", "playerteamgtype"]: run_select_fields.append(StratPlay.runner_team) if group_by in ["playergame", "teamgame"]: run_select_fields.append(StratPlay.game) run_plays = StratPlay.select(*run_select_fields).where( (StratPlay.game << season_games) & (StratPlay.runner.is_null(False)) ) if player_id is not None: all_players = Player.select().where(Player.player_id << player_id) bat_plays = bat_plays.where(StratPlay.batter << all_players) run_plays = run_plays.where(StratPlay.runner << all_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) bat_plays = bat_plays.where(StratPlay.batter_team << all_teams) run_plays = run_plays.where(StratPlay.runner_team << all_teams) if position is not None: bat_plays = bat_plays.where(StratPlay.batter_pos << position) if obc is not None: bat_plays = bat_plays.where(StratPlay.on_base_code << obc) if risp is not None: bat_plays = bat_plays.where( StratPlay.on_base_code << ["100", "101", "110", "111", "010", "011"] ) if inning is not None: bat_plays = bat_plays.where(StratPlay.inning_num << inning) if game_type is not None: all_types = [x.lower() for x in game_type] all_games = StratGame.select().where(fn.Lower(StratGame.game_type) << all_types) bat_plays = bat_plays.where(StratPlay.game << all_games) run_plays = run_plays.where(StratPlay.game << all_games) if is_gauntlet is not None: all_games = StratGame.select().where( fn.Lower(StratGame.game_type).contains("gauntlet") ) bat_plays = bat_plays.where(StratPlay.game << all_games) if group_by is not None: if group_by == "player": bat_plays = bat_plays.group_by(StratPlay.batter) run_plays = run_plays.group_by(StratPlay.runner) elif group_by == "team": bat_plays = bat_plays.group_by(StratPlay.batter_team) run_plays = run_plays.group_by(StratPlay.runner_team) elif group_by == "playerteam": bat_plays = bat_plays.group_by(StratPlay.batter, StratPlay.batter_team) run_plays = run_plays.group_by(StratPlay.runner, StratPlay.runner_team) elif group_by == "playergame": bat_plays = bat_plays.group_by(StratPlay.batter, StratPlay.game) run_plays = run_plays.group_by(StratPlay.runner, StratPlay.game) elif group_by == "teamgame": bat_plays = bat_plays.group_by(StratPlay.batter_team, StratPlay.game) run_plays = run_plays.group_by(StratPlay.runner_team, StratPlay.game) elif group_by == "league": bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by(StratPlay.game.season) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by(StratPlay.game.season) elif group_by == "gtype": bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by(StratPlay.game.game_type) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by(StratPlay.game.game_type) elif group_by == "playergtype": bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by(StratPlay.batter, StratPlay.game.game_type) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by(StratPlay.runner, StratPlay.game.game_type) elif group_by == "playerteamgtype": bat_plays = bat_plays.join(StratGame) bat_plays = bat_plays.group_by( StratPlay.batter, StratPlay.batter_team, StratPlay.game.game_type ) run_plays = run_plays.join(StratGame) run_plays = run_plays.group_by( StratPlay.runner, StratPlay.runner_team, StratPlay.game.game_type ) if sort is not None: if sort == "player": bat_plays = bat_plays.order_by(StratPlay.batter) run_plays = run_plays.order_by(StratPlay.runner) elif sort == "team": bat_plays = bat_plays.order_by(StratPlay.batter_team) run_plays = run_plays.order_by(StratPlay.runner_team) elif sort == "wpa-desc": bat_plays = bat_plays.order_by(SQL("sum_wpa").desc()) elif sort == "wpa-asc": bat_plays = bat_plays.order_by(SQL("sum_wpa").asc()) elif sort == "pa-desc": bat_plays = bat_plays.order_by(SQL("sum_pa").desc()) elif sort == "pa-asc": bat_plays = bat_plays.order_by(SQL("sum_pa").asc()) elif sort == "re24-desc": bat_plays = bat_plays.order_by(SQL("sum_re24").desc()) elif sort == "re24-asc": bat_plays = bat_plays.order_by(SQL("sum_re24").asc()) # NOTE: "newest" and "oldest" sort removed for GROUP BY queries # These require non-aggregated columns which PostgreSQL doesn't allow # elif sort == "newest": # bat_plays = bat_plays.order_by( # StratPlay.game_id.desc(), StratPlay.play_num.desc() # ) # run_plays = run_plays.order_by( # StratPlay.game_id.desc(), StratPlay.play_num.desc() # ) # elif sort == "oldest": # bat_plays = bat_plays.order_by(StratPlay.game_id, StratPlay.play_num) # run_plays = run_plays.order_by(StratPlay.game_id, StratPlay.play_num) if limit < 1: limit = 1 elif limit > 500: limit = 500 bat_plays = bat_plays.paginate(page_num, limit) logging.debug(f"bat_plays query: {bat_plays}") logging.debug(f"run_plays query: {run_plays}") # Convert to list first - .count() doesn't work on grouped queries in PostgreSQL bat_plays_list = list(bat_plays) return_stats = {"count": len(bat_plays_list), "stats": []} for x in bat_plays_list: # NOTE: Removed .order_by(StratPlay.id) - not valid with GROUP BY in PostgreSQL # and not meaningful for aggregated results anyway this_run = run_plays if "player" in group_by: this_run = this_run.where(StratPlay.runner == x.batter) if "game" in group_by: this_run = this_run.where(StratPlay.game == x.game) if "team" in group_by: this_run = this_run.where(StratPlay.runner_team == x.batter_team) if "gtype" in group_by: this_run = this_run.where(StratPlay.game.game_type == x.game.game_type) if this_run.count() > 0: sum_sb = this_run[0].sum_sb sum_cs = this_run[0].sum_cs run_wpa = this_run[0].sum_wpa # run_re24 = this_run[0].sum_re24 run_re24 = 0 else: sum_sb = 0 sum_cs = 0 run_wpa = 0 run_re24 = 0 this_wpa = bat_plays.where( (StratPlay.wpa >= min_wpa) & (StratPlay.wpa <= max_wpa) & (StratPlay.batter == x.batter) ) if this_wpa.count() > 0: sum_wpa = this_wpa[0].sum_wpa else: sum_wpa = 0 tot_ab = x.sum_ab if x.sum_ab > 0 else 1 obp = (x.sum_hit + x.sum_bb + x.sum_hbp + x.sum_ibb) / x.sum_pa slg = ( x.sum_hr * 4 + x.sum_triple * 3 + x.sum_double * 2 + (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) ) / tot_ab this_game = "TOT" if group_by in ["playergame", "teamgame"]: this_game = ( x.game_id if short_output else model_to_dict(x.game, recurse=False) ) elif "gtype" in group_by: this_game = x.game.game_type lob_all_rate, lob_2outs_rate, rbi_rate = 0, 0, 0 if x.count_runner1 + x.count_runner2 + x.count_runner3 > 0: lob_all_rate = (x.count_lo1 + x.count_lo2 + x.count_lo3) / ( x.count_runner1 + x.count_runner2 + x.count_runner3 ) rbi_rate = (x.sum_rbi - x.sum_hr) / ( x.count_runner1 + x.count_runner2 + x.count_runner3 ) return_stats["stats"].append( { "player": x.batter_id if short_output else ( model_to_dict(x.batter, recurse=True, max_depth=1) if x.batter_id else None ), "team": x.batter_team_id if short_output else ( model_to_dict(x.batter_team, recurse=True, max_depth=1) if x.batter_team_id else None ), "pa": x.sum_pa, "ab": x.sum_ab, "run": x.sum_run, "hit": x.sum_hit, "rbi": x.sum_rbi, "double": x.sum_double, "triple": x.sum_triple, "hr": x.sum_hr, "bb": x.sum_bb, "so": x.sum_so, "hbp": x.sum_hbp, "sac": x.sum_sac, "ibb": x.sum_ibb, "gidp": x.sum_gidp, "sb": sum_sb, "cs": sum_cs, "bphr": x.sum_bphr, "bpfo": x.sum_bpfo, "bp1b": x.sum_bp1b, "bplo": x.sum_bplo, "wpa": sum_wpa + run_wpa, "re24": x.sum_re24 + run_re24, "avg": x.sum_hit / tot_ab, "obp": obp, "slg": slg, "ops": obp + slg, "woba": ( 0.69 * x.sum_bb + 0.72 * x.sum_hbp + 0.89 * (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) + 1.27 * x.sum_double + 1.62 * x.sum_triple + 2.1 * x.sum_hr ) / max(x.sum_pa - x.sum_ibb, 1), "game": this_game, "lob_all": x.count_lo1 + x.count_lo2 + x.count_lo3, "lob_all_rate": lob_all_rate, "lob_2outs": x.count_lo1_3out + x.count_lo2_3out + x.count_lo3_3out, "rbi%": rbi_rate, } ) # if group_by == 're24-desc': # return_stats['stats'].sort(key=lambda x: x['re24'], reverse=True) # elif group_by == 're24-asc': # return_stats['stats'].sort(key=lambda x: x['re24']) if csv: return_vals = return_stats["stats"] if len(return_vals) == 0: return Response( content=pd.DataFrame().to_csv(index=False), media_type="text/csv" ) for x in return_vals: x["player_id"] = x["player"]["player_id"] if x["player"] else None x["player_name"] = x["player"]["p_name"] if x["player"] else None x["player_cardset"] = ( x["player"]["cardset"]["name"] if x["player"] and x["player"].get("cardset") else None ) x["team_id"] = x["team"]["id"] if x["team"] else None x["team_abbrev"] = x["team"]["abbrev"] if x["team"] else None if x.get("game") and isinstance(x["game"], dict) and "id" in x["game"]: x["game_id"] = x["game"]["id"] if "game_type" in x["game"]: x["game_type"] = x["game"]["game_type"] del x["game"] if x.get("player"): del x["player"] if x.get("team"): del x["team"] output = pd.DataFrame(return_vals) first = ["player_id", "player_name", "player_cardset", "team_id", "team_abbrev"] exclude = first + ["lob_all", "lob_all_rate", "lob_2outs", "rbi%"] output = output[first + [col for col in output.columns if col not in exclude]] db.close() return Response( content=pd.DataFrame(output).to_csv(index=False), media_type="text/csv" ) db.close() return return_stats @router.get("/pitching") async def get_pitching_totals( season: list = Query(default=None), week: list = Query(default=None), s_type: Literal["regular", "post", "total", None] = None, player_id: list = Query(default=None), group_by: Literal[ "team", "player", "playerteam", "playergame", "teamgame", "league", "gmtype", "playergtype", "playerteamgtype", ] = "player", is_gauntlet: Optional[bool] = None, min_pa: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), obc: list = Query(default=None), risp: Optional[bool] = None, inning: list = Query(default=None), page_num: Optional[int] = 1, game_type: list = Query(default=None), sort: Optional[str] = "ip-desc", limit: Optional[int] = 500, short_output: Optional[bool] = False, csv: Optional[bool] = False, ): season_games = StratGame.select() if season is not None: season_games = season_games.where(StratGame.season << season) if week is not None and s_type is not None: raise HTTPException( status_code=400, detail=f"Week and s_type parameters cannot be used in the same query", ) if week is not None: season_games = season_games.where(StratGame.week << week) if s_type is not None: if s_type == "regular": season_games = season_games.where(StratGame.week <= 18) elif s_type == "post": season_games = season_games.where(StratGame.week > 18) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) # Build SELECT fields conditionally based on group_by to satisfy PostgreSQL's # strict GROUP BY requirement (all non-aggregated SELECT fields must be in GROUP BY) pit_select_fields = [ fn.SUM(StratPlay.pa).alias("sum_pa"), fn.SUM(StratPlay.ab).alias("sum_ab"), fn.SUM(StratPlay.run).alias("sum_run"), fn.SUM(StratPlay.hit).alias("sum_hit"), fn.SUM(StratPlay.rbi).alias("sum_rbi"), fn.SUM(StratPlay.double).alias("sum_double"), fn.SUM(StratPlay.triple).alias("sum_triple"), fn.SUM(StratPlay.homerun).alias("sum_hr"), fn.SUM(StratPlay.bb).alias("sum_bb"), fn.SUM(StratPlay.so).alias("sum_so"), fn.SUM(StratPlay.wpa).alias("sum_wpa"), fn.SUM(StratPlay.hbp).alias("sum_hbp"), fn.SUM(StratPlay.sac).alias("sum_sac"), fn.SUM(StratPlay.ibb).alias("sum_ibb"), fn.SUM(StratPlay.gidp).alias("sum_gidp"), fn.SUM(StratPlay.sb).alias("sum_sb"), fn.SUM(StratPlay.cs).alias("sum_cs"), fn.SUM(StratPlay.bphr).alias("sum_bphr"), fn.SUM(StratPlay.bpfo).alias("sum_bpfo"), fn.SUM(StratPlay.bp1b).alias("sum_bp1b"), fn.SUM(StratPlay.bplo).alias("sum_bplo"), fn.SUM(StratPlay.wild_pitch).alias("sum_wp"), fn.SUM(StratPlay.balk).alias("sum_balk"), fn.SUM(StratPlay.outs).alias("sum_outs"), fn.SUM(StratPlay.e_run).alias("sum_erun"), fn.SUM(StratPlay.re24).alias("sum_re24"), fn.COUNT(StratPlay.on_first_final) .filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4) ) .alias("count_lo1"), fn.COUNT(StratPlay.on_second_final) .filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4) ) .alias("count_lo2"), fn.COUNT(StratPlay.on_third_final) .filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4) ) .alias("count_lo3"), fn.COUNT(StratPlay.on_first) .filter(StratPlay.on_first.is_null(False)) .alias("count_runner1"), fn.COUNT(StratPlay.on_second) .filter(StratPlay.on_second.is_null(False)) .alias("count_runner2"), fn.COUNT(StratPlay.on_third) .filter(StratPlay.on_third.is_null(False)) .alias("count_runner3"), fn.COUNT(StratPlay.on_first_final) .filter( StratPlay.on_first_final.is_null(False) & (StratPlay.on_first_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3) ) .alias("count_lo1_3out"), fn.COUNT(StratPlay.on_second_final) .filter( StratPlay.on_second_final.is_null(False) & (StratPlay.on_second_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3) ) .alias("count_lo2_3out"), fn.COUNT(StratPlay.on_third_final) .filter( StratPlay.on_third_final.is_null(False) & (StratPlay.on_third_final != 4) & (StratPlay.starting_outs + StratPlay.outs == 3) ) .alias("count_lo3_3out"), ] # Add non-aggregated fields based on grouping type if group_by in [ "player", "playerteam", "playergame", "playergtype", "playerteamgtype", ]: pit_select_fields.insert(0, StratPlay.pitcher) if group_by in ["team", "playerteam", "teamgame", "playerteamgtype"]: pit_select_fields.append(StratPlay.pitcher_team) if group_by in ["playergame", "teamgame"]: pit_select_fields.append(StratPlay.game) pit_plays = ( StratPlay.select(*pit_select_fields) .where((StratPlay.game << season_games) & (StratPlay.pitcher.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_pa) ) all_dec = ( Decision.select( Decision.pitcher, fn.SUM(Decision.win).alias("sum_win"), fn.SUM(Decision.loss).alias("sum_loss"), fn.SUM(Decision.hold).alias("sum_hold"), fn.SUM(Decision.is_save).alias("sum_save"), fn.SUM(Decision.b_save).alias("sum_b_save"), fn.SUM(Decision.irunners).alias("sum_irunners"), fn.SUM(Decision.irunners_scored).alias("sum_irun_scored"), # Cast boolean to integer for PostgreSQL compatibility fn.SUM(Case(None, [(Decision.is_start == True, 1)], 0)).alias("sum_gs"), fn.COUNT(Decision.game).alias("sum_game"), ) .where(Decision.game << season_games) .group_by(Decision.pitcher) ) if player_id is not None: all_players = Player.select().where(Player.player_id << player_id) pit_plays = pit_plays.where(StratPlay.pitcher << all_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) pit_plays = pit_plays.where(StratPlay.pitcher_team << all_teams) all_dec = all_dec.where(Decision.pitcher_team << all_teams) if obc is not None: pit_plays = pit_plays.where(StratPlay.on_base_code << obc) if risp is not None: pit_plays = pit_plays.where( StratPlay.on_base_code << ["100", "101", "110", "111", "010", "011"] ) if inning is not None: pit_plays = pit_plays.where(StratPlay.inning_num << inning) if game_type is not None: all_types = [x.lower() for x in game_type] all_games = StratGame.select().where(fn.Lower(StratGame.game_type) << all_types) pit_plays = pit_plays.where(StratPlay.game << all_games) if group_by is not None: if group_by == "player": pit_plays = pit_plays.group_by(StratPlay.pitcher) elif group_by == "team": pit_plays = pit_plays.group_by(StratPlay.pitcher_team) elif group_by == "playerteam": pit_plays = pit_plays.group_by(StratPlay.pitcher, StratPlay.pitcher_team) elif group_by == "playergame": pit_plays = pit_plays.group_by(StratPlay.pitcher, StratPlay.game) elif group_by == "teamgame": pit_plays = pit_plays.group_by(StratPlay.pitcher_team, StratPlay.game) elif group_by == "league": pit_plays = pit_plays.join(StratGame) pit_plays = pit_plays.group_by(StratPlay.game.season) elif group_by == "gtype": pit_plays = pit_plays.join(StratGame) pit_plays = pit_plays.group_by(StratPlay.game.game_type) elif group_by == "playergtype": pit_plays = pit_plays.join(StratGame) pit_plays = pit_plays.group_by(StratPlay.pitcher, StratPlay.game.game_type) elif group_by == "playerteamgtype": pit_plays = pit_plays.join(StratGame) pit_plays = pit_plays.group_by( StratPlay.pitcher, StratPlay.pitcher_team, StratPlay.game.game_type ) if sort is not None: if sort == "player": pit_plays = pit_plays.order_by(StratPlay.pitcher) elif sort == "team": pit_plays = pit_plays.order_by(StratPlay.pitcher_team) elif sort == "wpa-desc": pit_plays = pit_plays.order_by( SQL("sum_wpa").asc() ) # functions seem reversed since pitcher plays negative elif sort == "wpa-asc": pit_plays = pit_plays.order_by(SQL("sum_wpa").desc()) elif sort == "re24-desc": pit_plays = pit_plays.order_by( SQL("sum_re24").asc() ) # functions seem reversed since pitcher plays negative elif sort == "re24-asc": pit_plays = pit_plays.order_by(SQL("sum_re24").desc()) elif sort == "ip-desc": pit_plays = pit_plays.order_by(SQL("sum_outs").desc()) elif sort == "ip-asc": pit_plays = pit_plays.order_by(SQL("sum_outs").asc()) elif sort == "game-desc": pit_plays = pit_plays.order_by(SQL("sum_game").desc()) elif sort == "game-asc": pit_plays = pit_plays.order_by(SQL("sum_game").asc()) # NOTE: "newest" and "oldest" sort removed for GROUP BY queries # These require non-aggregated columns which PostgreSQL doesn't allow # elif sort == "newest": # pit_plays = pit_plays.order_by( # StratPlay.game_id.desc(), StratPlay.play_num.desc() # ) # elif sort == "oldest": # pit_plays = pit_plays.order_by(StratPlay.game_id, StratPlay.play_num) if limit < 1: limit = 1 elif limit > 500: limit = 500 pit_plays = pit_plays.paginate(page_num, limit) # Convert to list first - .count() doesn't work on grouped queries in PostgreSQL pit_plays_list = list(pit_plays) return_stats = {"count": len(pit_plays_list), "stats": []} for x in pit_plays_list: this_dec = all_dec.where(Decision.pitcher == x.pitcher) if game_type is not None: all_types = [x.lower() for x in game_type] all_games = StratGame.select().where( fn.Lower(StratGame.game_type) << all_types ) this_dec = this_dec.where(Decision.game << all_games) tot_outs = x.sum_outs if x.sum_outs > 0 else 1 obp = (x.sum_hit + x.sum_bb + x.sum_hbp + x.sum_ibb) / x.sum_pa slg = ( x.sum_hr * 4 + x.sum_triple * 3 + x.sum_double * 2 + (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) ) / max(x.sum_ab, 1) tot_bb = 0.1 if x.sum_bb == 0 else x.sum_bb this_game = "TOT" if group_by in ["playergame", "teamgame"]: this_game = ( x.game_id if short_output else model_to_dict(x.game, recurse=False) ) this_dec = all_dec.where( (Decision.pitcher == x.pitcher) & (Decision.game == x.game) ) elif "gtype" in group_by: this_game = x.game.game_type lob_all_rate, lob_2outs_rate, rbi_rate = 0, 0, 0 if x.count_runner1 + x.count_runner2 + x.count_runner3 > 0: lob_all_rate = (x.count_lo1 + x.count_lo2 + x.count_lo3) / ( x.count_runner1 + x.count_runner2 + x.count_runner3 ) rbi_rate = (x.sum_rbi - x.sum_hr) / ( x.count_runner1 + x.count_runner2 + x.count_runner3 ) return_stats["stats"].append( { "player": x.pitcher_id if short_output else (model_to_dict(x.pitcher) if x.pitcher_id else None), "team": x.pitcher_team_id if short_output else (model_to_dict(x.pitcher_team) if x.pitcher_team_id else None), "tbf": x.sum_pa, "outs": x.sum_outs, "games": this_dec[0].sum_game, "gs": this_dec[0].sum_gs, "win": this_dec[0].sum_win, "loss": this_dec[0].sum_loss, "hold": this_dec[0].sum_hold, "save": this_dec[0].sum_save, "bsave": this_dec[0].sum_b_save, "ir": this_dec[0].sum_irunners, "ir_sc": this_dec[0].sum_irun_scored, "ab": x.sum_ab, "run": x.sum_run, "e_run": x.sum_erun, "hits": x.sum_hit, "double": x.sum_double, "triple": x.sum_triple, "hr": x.sum_hr, "bb": x.sum_bb, "so": x.sum_so, "hbp": x.sum_hbp, "sac": x.sum_sac, "ibb": x.sum_ibb, "gidp": x.sum_gidp, "sb": x.sum_sb, "cs": x.sum_cs, "bphr": x.sum_bphr, "bpfo": x.sum_bpfo, "bp1b": x.sum_bp1b, "bplo": x.sum_bplo, "wp": x.sum_wp, "balk": x.sum_balk, "wpa": x.sum_wpa * -1, "re24": x.sum_re24 * -1, "era": (x.sum_erun * 27) / tot_outs, "whip": ((x.sum_bb + x.sum_hit + x.sum_ibb) * 3) / tot_outs, "avg": x.sum_hit / max(x.sum_ab, 1), "obp": obp, "slg": slg, "ops": obp + slg, "woba": ( 0.69 * x.sum_bb + 0.72 * x.sum_hbp + 0.89 * (x.sum_hit - x.sum_double - x.sum_triple - x.sum_hr) + 1.27 * x.sum_double + 1.62 * x.sum_triple + 2.1 * x.sum_hr ) / max(x.sum_pa - x.sum_ibb, 1), "k/9": x.sum_so * 9 / (tot_outs / 3), "bb/9": x.sum_bb * 9 / (tot_outs / 3), "k/bb": x.sum_so / tot_bb, "game": this_game, "lob_2outs": x.count_lo1_3out + x.count_lo2_3out + x.count_lo3_3out, "rbi%": rbi_rate, } ) db.close() if csv: return_vals = return_stats["stats"] if len(return_vals) == 0: return Response( content=pd.DataFrame().to_csv(index=False), media_type="text/csv" ) for x in return_vals: x["player_id"] = x["player"]["player_id"] if x["player"] else None x["player_name"] = x["player"]["p_name"] if x["player"] else None x["player_cardset"] = ( x["player"]["cardset"]["name"] if x["player"] and x["player"].get("cardset") else None ) x["team_id"] = x["team"]["id"] if x["team"] else None x["team_abbrev"] = x["team"]["abbrev"] if x["team"] else None if x.get("game") and isinstance(x["game"], dict) and "id" in x["game"]: x["game_id"] = x["game"]["id"] if "game_type" in x["game"]: x["game_type"] = x["game"]["game_type"] del x["game"] if x.get("player"): del x["player"] if x.get("team"): del x["team"] output = pd.DataFrame(return_vals) first = ["player_id", "player_name", "player_cardset", "team_id", "team_abbrev"] exclude = first + ["lob_2outs", "rbi%"] output = output[first + [col for col in output.columns if col not in exclude]] db.close() return Response( content=pd.DataFrame(output).to_csv(index=False), media_type="text/csv" ) return return_stats @router.get("/game-summary/{game_id}") async def get_game_summary( game_id: int, csv: Optional[bool] = False, short_output: Optional[bool] = False, tp_max: Optional[int] = 1, poop_max: Optional[int] = 1, ): this_game = StratGame.get_or_none(StratGame.id == game_id) if this_game is None: db.close() raise HTTPException(status_code=404, detail=f"Game {game_id} not found") game_plays = StratPlay.select().where(StratPlay.game_id == game_id) all_hits = game_plays.where(StratPlay.hit == 1) all_errors = game_plays.where(StratPlay.error == 1) all_runs = game_plays.where(StratPlay.run == 1) all_dec = Decision.select().where(Decision.game_id == game_id) winner = all_dec.where(Decision.win == 1) loser = all_dec.where(Decision.loss == 1) save_p = all_dec.where(Decision.is_save == 1) save_pitcher = None if save_p.count() > 0: save_pitcher = model_to_dict(save_p.get().pitcher, recurse=not short_output) all_holds = all_dec.where(Decision.hold == 1) all_bsaves = all_dec.where(Decision.b_save == 1) doubles = all_hits.where(StratPlay.double == 1) triples = all_hits.where(StratPlay.triple == 1) homeruns = all_hits.where((StratPlay.homerun == 1) | (StratPlay.bphr == 1)) steal_att = game_plays.where(StratPlay.runner.is_null(False)) all_sb = steal_att.where(StratPlay.sb == 1) all_cs = steal_att.where(StratPlay.cs == 1) all_batters = ( StratPlay.select( StratPlay.batter, fn.SUM(StratPlay.re24).alias("sum_re24"), fn.SUM(StratPlay.ab).alias("sum_ab"), fn.SUM(StratPlay.run).alias("sum_run"), fn.SUM(StratPlay.hit).alias("sum_hit"), fn.SUM(StratPlay.rbi).alias("sum_rbi"), fn.SUM(StratPlay.double).alias("sum_double"), fn.SUM(StratPlay.triple).alias("sum_triple"), fn.SUM(StratPlay.homerun).alias("sum_hr"), fn.SUM(StratPlay.bphr).alias("sum_bphr"), ) .where(StratPlay.game_id == game_id) .group_by(StratPlay.batter, StratPlay.batter_team) ) all_pitchers = ( StratPlay.select( StratPlay.pitcher, fn.SUM(StratPlay.re24).alias("sum_re24"), fn.SUM(StratPlay.pa).alias("sum_pa"), fn.SUM(StratPlay.outs).alias("sum_outs"), fn.SUM(StratPlay.e_run).alias("sum_erun"), fn.SUM(StratPlay.run).alias("sum_run"), fn.SUM(StratPlay.so).alias("sum_so"), fn.SUM(StratPlay.hit).alias("sum_hit"), ) .where(StratPlay.game_id == game_id) .group_by(StratPlay.pitcher, StratPlay.pitcher_team) ) top_pitchers = all_pitchers.order_by(SQL("sum_re24").asc()).limit(tp_max) top_batters = all_batters.order_by(SQL("sum_re24").desc()).limit(tp_max) bot_pitcher = all_pitchers.order_by(SQL("sum_re24").desc()).get() bot_batter = all_batters.order_by(SQL("sum_re24").asc()).get() top_b = [ { "player": model_to_dict(x.batter, recurse=not short_output), "ab": x.sum_ab, "run": x.sum_run, "hit": x.sum_hit, "rbi": x.sum_rbi, "double": x.sum_double, "triple": x.sum_triple, "hr": x.sum_hr, "re24": x.sum_re24, } for x in top_batters ] top_p = [ { "player": model_to_dict(x.pitcher, recurse=not short_output), "tbf": x.sum_pa, "ip": math.floor(x.sum_outs / 3) + ((x.sum_outs % 3) * 0.1), "run": x.sum_run, "e_run": x.sum_erun, "hit": x.sum_hit, "so": x.sum_so, "re24": x.sum_re24 * -1, } for x in top_pitchers ] top_players = [*top_b, *top_p] logging.debug(f"top_players: {top_players}") bot_players = [ { "player": model_to_dict(bot_pitcher.pitcher, recurse=not short_output), "tbf": bot_pitcher.sum_pa, "ip": math.floor(bot_pitcher.sum_outs / 3) + ((bot_pitcher.sum_outs % 3) * 0.1), "run": bot_pitcher.sum_run, "e_run": bot_pitcher.sum_erun, "hit": bot_pitcher.sum_hit, "so": bot_pitcher.sum_so, "re24": bot_pitcher.sum_re24 * -1, }, { "player": model_to_dict(bot_batter.batter, recurse=not short_output), "ab": bot_batter.sum_ab, "run": bot_batter.sum_run, "hit": bot_batter.sum_hit, "rbi": bot_batter.sum_rbi, "double": bot_batter.sum_double, "triple": bot_batter.sum_triple, "hr": bot_batter.sum_hr, "re24": bot_batter.sum_re24, }, ] return { "game": model_to_dict(this_game, recurse=not short_output), "teams": { "away": model_to_dict(this_game.away_team, recurse=not short_output), "home": model_to_dict(this_game.home_team, recurse=not short_output), }, "runs": { "away": all_runs.where( StratPlay.batter_team == this_game.away_team ).count(), "home": all_runs.where( StratPlay.batter_team == this_game.home_team ).count(), }, "hits": { "away": all_hits.where( StratPlay.batter_team == this_game.away_team ).count(), "home": all_hits.where( StratPlay.batter_team == this_game.home_team ).count(), }, "errors": { "away": all_errors.where( StratPlay.defender_team == this_game.away_team ).count(), "home": all_errors.where( StratPlay.defender_team == this_game.home_team ).count(), }, "top-players": sorted(top_players, key=lambda x: x["re24"], reverse=True)[ :tp_max ], "pooper": sorted(bot_players, key=lambda x: x["re24"])[:poop_max], "pitchers": { "win": model_to_dict(winner.get().pitcher, recurse=not short_output), "loss": model_to_dict(loser.get().pitcher, recurse=not short_output), "holds": [ model_to_dict(x.pitcher, recurse=not short_output) for x in all_holds ], "save": save_pitcher, "b_saves": [ model_to_dict(x.pitcher, recurse=not short_output) for x in all_bsaves ], }, "xbh": { "2b": [model_to_dict(x.batter, recurse=not short_output) for x in doubles], "3b": [model_to_dict(x.batter, recurse=not short_output) for x in triples], "hr": [model_to_dict(x.batter, recurse=not short_output) for x in homeruns], }, "running": { "sb": [model_to_dict(x.runner, recurse=not short_output) for x in all_sb], "csc": [model_to_dict(x.catcher, recurse=not short_output) for x in all_cs], }, } @router.get("/{play_id}") async def get_one_play(play_id: int): if StratPlay.get_or_none(StratPlay.id == play_id) is None: db.close() raise HTTPException(status_code=404, detail=f"Play ID {play_id} not found") r_play = model_to_dict(StratPlay.get_by_id(play_id)) db.close() return r_play @router.patch("/{play_id}") async def patch_play( play_id: int, new_play: PlayModel, token: str = Depends(oauth2_scheme) ): if not valid_token(token): logging.warning(f"patch_play - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") if StratPlay.get_or_none(StratPlay.id == play_id) is None: db.close() raise HTTPException(status_code=404, detail=f"Play ID {play_id} not found") StratPlay.update(**new_play.dict()).where(StratPlay.id == play_id).execute() r_play = model_to_dict(StratPlay.get_by_id(play_id)) db.close() return r_play @router.post("") async def post_plays(p_list: PlayList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"post_plays - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") new_plays = [] this_game = StratGame.get_or_none(StratGame.id == p_list.plays[0].game_id) if this_game is None: raise HTTPException( status_code=404, detail=f"Game ID {p_list.plays[0].game_id} not found" ) for play in p_list.plays: this_play = play this_play.inning_half = this_play.inning_half.lower() top_half = this_play.inning_half == "top" if this_play.batter_team_id is None and this_play.batter_id is not None: this_play.batter_team_id = ( this_game.away_team.id if top_half else this_game.home_team.id ) if this_play.pitcher_team_id is None: this_play.pitcher_team_id = ( this_game.home_team.id if top_half else this_game.away_team.id ) if this_play.catcher_id is not None: this_play.catcher_team_id = ( this_game.home_team.id if top_half else this_game.away_team.id ) if this_play.defender_id is not None: this_play.defender_team_id = ( this_game.home_team.id if top_half else this_game.away_team.id ) if this_play.runner_id is not None: this_play.runner_team_id = ( this_game.away_team.id if top_half else this_game.home_team.id ) if this_play.pa == 0: this_play.batter_final = None new_plays.append(this_play.dict()) with db.atomic(): # Use PostgreSQL-compatible upsert helper upsert_strat_plays(new_plays, batch_size=20) db.close() return f"Inserted {len(new_plays)} plays" @router.delete("/{play_id}") async def delete_play(play_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"delete_play - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") this_play = StratPlay.get_or_none(StratPlay.id == play_id) if not this_play: db.close() raise HTTPException(status_code=404, detail=f"Play ID {play_id} not found") count = this_play.delete_instance() db.close() if count == 1: return f"Play {play_id} has been deleted" else: raise HTTPException( status_code=500, detail=f"Play {play_id} could not be deleted" ) @router.delete("/game/{game_id}") async def delete_plays_game(game_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"delete_plays_game - Bad Token: {token}") raise HTTPException(status_code=401, detail="Unauthorized") this_game = StratGame.get_or_none(StratGame.id == game_id) if not this_game: db.close() raise HTTPException(status_code=404, detail=f"Game ID {game_id} not found") count = StratPlay.delete().where(StratPlay.game == this_game).execute() db.close() if count > 0: return f"Deleted {count} plays matching Game ID {game_id}" else: raise HTTPException( status_code=500, detail=f"No plays matching Game ID {game_id} were deleted" )