from fastapi import APIRouter, Depends, HTTPException, Query from typing import List, Optional, Literal import copy import logging import pydantic from ..db_engine import db, StratPlay, StratGame, Team, model_to_dict, chunked, fn 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(pydantic.BaseModel): game_id: int play_num: int batter_id: int batter_team_id: int = None pitcher_id: int pitcher_team_id: int = None on_base_code: str inning_half: Literal['top', 'bot'] inning_num: int batting_order: int starting_outs: int away_score: int home_score: int batter_pos: POS_LIST 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 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 class PlayList(pydantic.BaseModel): plays: List[PlayModel] @router.get('') async def get_plays( game_id: list = Query(default=None), batter_id: 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), short_output: Optional[bool] = False, 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 limit > 5000: limit = 5000 elif limit < 1: limit = 1 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.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 top_half = this_play.inning_half == 'top' if this_play.batter_team_id is 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 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')