from fastapi import APIRouter, Depends, HTTPException, Query from typing import List, Optional, Literal import copy import logging from pydantic import BaseModel, validator from ..db_engine import db, StratPlay, StratGame, Team, Player, Decision, model_to_dict, chunked, fn, SQL 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/v3/plays', tags=['plays'] ) POS_LIST = Literal['C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF', 'P', 'DH', 'PH', 'PR'] 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 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('') # Want to add runner parameters 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), 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), double: Optional[int] = None, triple: Optional[int] = None, homerun: Optional[int] = None, sb: Optional[int] = None, cs: Optional[int] = None, manager_id: list = Query(default=None), 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, short_output: Optional[bool] = False, sort: Optional[str] = None, limit: Optional[int] = 200): all_plays = StratPlay.select() 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.pitcher_team << all_teams) | (StratPlay.catcher_team << all_teams) | (StratPlay.defender_team << all_teams) ) 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 manager_id is not None: all_games = StratGame.select().where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) all_plays = all_plays.where(StratPlay.game << all_games) 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 limit > 5000: limit = 5000 elif limit < 1: limit = 1 if sort == 'wpa-desc': all_plays = all_plays.order_by(-fn.ABS(StratPlay.wpa)) if sort == 'wpa-asc': all_plays = all_plays.order_by(fn.ABS(StratPlay.wpa)) all_plays = all_plays.limit(limit) 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), s_type: Literal['regular', 'post', 'total', None] = None, position: list = Query(default=None), player_id: list = Query(default=None), group_by: Literal['team', 'player', 'playerteam'] = 'player', min_pa: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), sort: Optional[str] = None, limit: Optional[int] = None, short_output: 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) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) bat_plays = ( StratPlay .select(StratPlay.batter, 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'), StratPlay.batter_team, 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')) .where((StratPlay.game << season_games) & (StratPlay.batter.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_pa) ) run_plays = ( StratPlay .select(StratPlay.runner, StratPlay.runner_team, 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')) .where((StratPlay.game << season_games) & (StratPlay.runner.is_null(False))) ) def_plays = ( StratPlay .select(StratPlay.defender, StratPlay.defender_team, fn.SUM(StratPlay.error).alias('sum_error'), fn.SUM(StratPlay.hit).alias('sum_hit'), fn.SUM(StratPlay.pa).alias('sum_chances'), fn.SUM(StratPlay.wpa).alias('sum_wpa')) .where((StratPlay.game << season_games) & (StratPlay.defender.is_null(False))) ) if player_id is not None: all_players = Player.select().where(Player.id << player_id) bat_plays = bat_plays.where(StratPlay.batter << all_players) run_plays = run_plays.where(StratPlay.runner << all_players) def_plays = def_plays.where(StratPlay.defender << 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) def_plays = def_plays.where(StratPlay.defender_team << all_teams) if position is not None: bat_plays = bat_plays.where(StratPlay.batter_pos << position) 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) def_plays = def_plays.group_by(StratPlay.defender) elif group_by == 'team': bat_plays = bat_plays.group_by(StratPlay.batter_team) run_plays = run_plays.group_by(StratPlay.runner_team) def_plays = def_plays.group_by(StratPlay.defender_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) def_plays = def_plays.group_by(StratPlay.defender, StratPlay.defender_team) if sort is not None: if sort == 'player': bat_plays = bat_plays.order_by(StratPlay.batter) run_plays = run_plays.order_by(StratPlay.runner) def_plays = def_plays.order_by(StratPlay.defender) elif sort == 'team': bat_plays = bat_plays.order_by(StratPlay.batter_team) run_plays = run_plays.order_by(StratPlay.runner_team) def_plays = def_plays.order_by(StratPlay.defender_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()) if limit is not None: if limit < 1: limit = 1 bat_plays = bat_plays.limit(limit) logging.info(f'bat_plays query: {bat_plays}') logging.info(f'run_plays query: {run_plays}') return_stats = { 'count': bat_plays.count(), 'stats': [] } for x in bat_plays: this_run = run_plays.where(StratPlay.runner == x.batter) if this_run.count() > 0: sum_sb = this_run[0].sum_sb sum_cs = this_run[0].sum_cs sum_wpa = this_run[0].sum_wpa else: sum_sb = 0 sum_cs = 0 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 return_stats['stats'].append({ 'player': x.batter_id if short_output else model_to_dict(x.batter, recurse=False), 'team': x.batter_team_id if short_output else model_to_dict(x.batter_team, recurse=False), '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': x.sum_wpa + sum_wpa, 'avg': x.sum_hit / tot_ab, 'obp': obp, 'slg': slg, 'ops': obp + slg, 'woba': (.69 * x.sum_bb + .72 * x.sum_hbp + .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) / (x.sum_pa - x.sum_ibb) }) # Get Running Stats # Get Fielding Stats 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'] = 'player', min_pa: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), sort: Optional[str] = None, limit: Optional[int] = None, short_output: 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) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) pit_plays = ( StratPlay .select(StratPlay.pitcher, StratPlay.pitcher_team, 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.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')) .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'), fn.SUM(Decision.is_start).alias('sum_gs'), fn.COUNT(Decision.game).alias('sum_game')) .where(Decision.game << season_games) ) if player_id is not None: all_players = Player.select().where(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) 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) 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 == '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()) if limit is not None: if limit < 1: limit = 1 pit_plays = pit_plays.limit(limit) return_stats = { 'count': pit_plays.count(), 'stats': [] } for x in pit_plays: this_dec = all_dec.where(Decision.pitcher == x.pitcher) 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)) / x.sum_ab return_stats['stats'].append({ 'player': x.pitcher_id if short_output else model_to_dict(x.pitcher, recurse=False), 'team': x.pitcher_team_id if short_output else model_to_dict(x.pitcher_team, recurse=False), '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, 'era': (x.sum_erun * 27) / tot_outs, 'whip': ((x.sum_bb + x.sum_hit + x.sum_ibb) * 3) / tot_outs, 'avg': x.sum_hit / x.sum_ab, 'obp': obp, 'slg': slg, 'ops': obp + slg, 'woba': (.69 * x.sum_bb + .72 * x.sum_hbp + .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) / (x.sum_pa - x.sum_ibb) }) db.close() return return_stats @router.get('/fielding') async def get_fielding_totals( season: list = Query(default=None), week: list = Query(default=None), s_type: Literal['regular', 'post', 'total', None] = None, position: list = Query(default=None), player_id: list = Query(default=None), group_by: Literal['team', 'player', 'playerteam'] = 'player', min_ch: Optional[int] = 1, team_id: list = Query(default=None), manager_id: list = Query(default=None), sort: Optional[str] = None, limit: Optional[int] = None, short_output: 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) if manager_id is not None: season_games = season_games.where( (StratGame.away_manager_id << manager_id) | (StratGame.home_manager_id << manager_id) ) def_plays = ( StratPlay .select(StratPlay.defender, StratPlay.defender_team, fn.SUM(StratPlay.error).alias('sum_error'), fn.SUM(StratPlay.hit).alias('sum_hit'), fn.SUM(StratPlay.pa).alias('sum_chances'), fn.SUM(StratPlay.wpa).alias('sum_wpa')) .where((StratPlay.game << season_games) & (StratPlay.defender.is_null(False))) .having(fn.SUM(StratPlay.pa) >= min_ch) ) cat_plays = ( StratPlay .select(StratPlay.catcher, StratPlay.catcher_team, fn.SUM(StratPlay.sb).alias('sum_sb'), fn.SUM(StratPlay.cs).alias('sum_cs'), fn.SUM(StratPlay.wpa).alias('sum_wpa'), fn.SUM(StratPlay.passed_ball).alias('sum_pb')) .where((StratPlay.game << season_games) & (StratPlay.catcher.is_null(False))) ) if player_id is not None: all_players = Player.select().where(Player.id << player_id) def_plays = def_plays.where(StratPlay.defender << all_players) cat_plays = cat_plays.where(StratPlay.catcher << all_players) if team_id is not None: all_teams = Team.select().where(Team.id << team_id) def_plays = def_plays.where(StratPlay.defender_team << all_teams) cat_plays = cat_plays.where(StratPlay.catcher_team << all_teams) if position is not None: def_plays = def_plays.where(StratPlay.check_pos << position) if group_by is not None: if group_by == 'player': def_plays = def_plays.group_by(StratPlay.defender) cat_plays = cat_plays.group_by(StratPlay.catcher) elif group_by == 'team': def_plays = def_plays.group_by(StratPlay.defender_team) cat_plays = cat_plays.group_by(StratPlay.catcher_team) elif group_by == 'playerteam': def_plays = def_plays.group_by(StratPlay.defender, StratPlay.defender_team) cat_plays = cat_plays.group_by(StratPlay.catcher, StratPlay.catcher_team) if sort is not None: if sort == 'player': def_plays = def_plays.order_by(StratPlay.defender) elif sort == 'team': def_plays = def_plays.order_by(StratPlay.defender_team) elif sort == 'wpa-desc': def_plays = def_plays.order_by(SQL('sum_wpa').asc()) elif sort == 'wpa-asc': def_plays = def_plays.order_by(SQL('sum_wpa').desc()) elif sort == 'ch-desc': def_plays = def_plays.order_by(SQL('sum_chances').desc()) elif sort == 'ch-asc': def_plays = def_plays.order_by(SQL('sum_chances').asc()) if limit is not None: if limit < 1: limit = 1 def_plays = def_plays.limit(limit) logging.info(f'def_plays query: {def_plays}') return_stats = { 'count': def_plays.count(), 'stats': [] } for x in def_plays: this_cat = cat_plays.where(StratPlay.catcher == x.defender) if this_cat.count() > 0: sum_sb = this_cat[0].sum_sb sum_cs = this_cat[0].sum_cs sum_wpa = this_cat[0].sum_wpa sum_pb = this_cat[0].sum_pb else: sum_sb = 0 sum_cs = 0 sum_wpa = 0 sum_pb = 0 return_stats['stats'].append({ 'player': x.defender_id if short_output else model_to_dict(x.defender, recurse=False), 'team': x.defender_team_id if short_output else model_to_dict(x.defender_team, recurse=False), 'pos': 'TOT' if position is None or len(position) > 1 else position[0], 'x-ch': x.sum_chances, 'hit': x.sum_hit, 'error': x.sum_error, 'sb-ch': sum_sb + sum_cs, 'sb': sum_sb, 'cs': sum_cs, 'pb': sum_pb, 'wpa': (x.sum_wpa + sum_wpa) * -1, 'wf%': (x.sum_chances - (x.sum_error * .5) - (x.sum_hit * .75)) / x.sum_chances, 'cs%': sum_cs / (sum_sb + sum_cs) if (sum_sb + sum_cs) > 0 else '' }) db.close() return return_stats @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(): for batch in chunked(new_plays, 20): StratPlay.insert_many(batch).on_conflict_replace().execute() 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')