from fastapi import APIRouter, Depends, HTTPException, Query, Response from typing import List, Optional, Literal import copy import logging import pandas as pd import pydantic from ..db_engine import db, Decision, StratGame, Player, model_to_dict, chunked, fn, Team 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/decisions', tags=['decisions'] ) class DecisionModel(pydantic.BaseModel): game_id: int season: int week: int pitcher_id: int pitcher_team_id: int win: int = 0 loss: int = 0 hold: int = 0 is_save: int = 0 is_start: bool = False b_save: int = 0 irunners: int = 0 irunners_scored: int = 0 rest_ip: float = 0 rest_required: int = 0 class DecisionList(pydantic.BaseModel): decisions: List[DecisionModel] @router.get('') async def get_decisions( season: list = Query(default=None), week: list = Query(default=None), team_id: list = Query(default=None), win: Optional[int] = None, loss: Optional[int] = None, hold: Optional[int] = None, save: Optional[int] = None, b_save: Optional[int] = None, irunners: list = Query(default=None), irunners_scored: list = Query(default=None), game_type: list = Query(default=None), game_id: list = Query(default=None), player_id: list = Query(default=None), csv: Optional[bool] = False, limit: Optional[int] = 100, page_num: Optional[int] = 1, short_output: Optional[bool] = False): all_dec = Decision.select().order_by(-Decision.season, -Decision.week, -Decision.id) if season is not None: all_dec = all_dec.where(Decision.season << season) if week is not None: all_dec = all_dec.where(Decision.week << week) if game_id is not None: all_dec = all_dec.where(Decision.game_id << game_id) if player_id is not None: all_dec = all_dec.where(Decision.pitcher_id << player_id) if team_id is not None: all_dec = all_dec.where(Decision.pitcher_team_id << team_id) if win is not None: all_dec = all_dec.where(Decision.win == win) if loss is not None: all_dec = all_dec.where(Decision.loss == loss) if hold is not None: all_dec = all_dec.where(Decision.hold == hold) if save is not None: all_dec = all_dec.where(Decision.save == save) if b_save is not None: all_dec = all_dec.where(Decision.b_save == b_save) if irunners is not None: all_dec = all_dec.where(Decision.irunners << irunners) if irunners_scored is not None: all_dec = all_dec.where(Decision.irunners_scored << irunners_scored) 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_dec = all_dec.where(Decision.game << all_games) if limit < 1: limit = 1 if limit > 100: limit = 100 all_dec = all_dec.paginate(page_num, limit) return_dec = { 'count': all_dec.count(), 'decisions': [model_to_dict(x, recurse=not short_output) for x in all_dec] } db.close() if csv: return_vals = return_dec['decisions'] if len(return_vals) == 0: return Response(content=pd.DataFrame().to_csv(index=False), media_type='text/csv') for x in return_vals: x['game_id'] = x['game']['id'] x['game_type'] = x['game']['game_type'] x['player_id'] = x['pitcher']['player_id'] x['player_name'] = x['pitcher']['p_name'] x['player_cardset'] = x['pitcher']['cardset']['name'] x['team_id'] = x['pitcher_team']['id'] x['team_abbrev'] = x['pitcher_team']['abbrev'] del x['pitcher'], x['pitcher_team'], x['game'] 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') return return_dec @router.patch('/{decision_id}') async def patch_decision( decision_id: int, win: Optional[int] = None, loss: Optional[int] = None, hold: Optional[int] = None, save: Optional[int] = None, b_save: Optional[int] = None, irunners: Optional[int] = None, irunners_scored: Optional[int] = None, rest_ip: Optional[int] = None, rest_required: Optional[int] = None, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'patch_decision - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') this_dec = Decision.get_or_none(Decision.id == decision_id) if this_dec is None: db.close() raise HTTPException(status_code=404, detail=f'Decision ID {decision_id} not found') if win is not None: this_dec.win = win if loss is not None: this_dec.loss = loss if hold is not None: this_dec.hold = hold if save is not None: this_dec.is_save = save if b_save is not None: this_dec.b_save = b_save if irunners is not None: this_dec.irunners = irunners if irunners_scored is not None: this_dec.irunners_scored = irunners_scored if rest_ip is not None: this_dec.rest_ip = rest_ip if rest_required is not None: this_dec.rest_required = rest_required if this_dec.save() == 1: d_result = model_to_dict(this_dec) db.close() return d_result else: db.close() raise HTTPException(status_code=500, detail=f'Unable to patch decision {decision_id}') @router.post('') async def post_decisions(dec_list: DecisionList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'post_decisions - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') new_dec = [] for x in dec_list.decisions: if StratGame.get_or_none(StratGame.id == x.game_id) is None: raise HTTPException(status_code=404, detail=f'Game ID {x.game_id} not found') if Player.get_or_none(Player.player_id == x.pitcher_id) is None: raise HTTPException(status_code=404, detail=f'Player ID {x.pitcher_id} not found') if Team.get_or_none(Team.id == x.pitcher_team_id) is None: raise HTTPException(status_code=404, detail=f'Team ID {x.pitcher_team_id} not found') new_dec.append(x.dict()) with db.atomic(): for batch in chunked(new_dec, 10): Decision.insert_many(batch).on_conflict_replace().execute() db.close() return f'Inserted {len(new_dec)} decisions' @router.delete('/{decision_id}') async def delete_decision(decision_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'delete_decision - Bad Token: {token}') raise HTTPException(status_code=401, detail='Unauthorized') this_dec = Decision.get_or_none(Decision.id == decision_id) if this_dec is None: db.close() raise HTTPException(status_code=404, detail=f'Decision ID {decision_id} not found') count = this_dec.delete_instance() db.close() if count == 1: return f'Decision {decision_id} has been deleted' else: raise HTTPException(status_code=500, detail=f'Decision {decision_id} could not be deleted') @router.delete('/game/{game_id}') async def delete_decisions_game(game_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'delete_decisions_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 = Decision.delete().where(Decision.game == this_game).execute() db.close() if count > 0: return f'Deleted {count} decisions matching Game ID {game_id}' else: raise HTTPException(status_code=500, detail=f'No decisions matching Game ID {game_id} were deleted')