import datetime import random from fastapi import APIRouter, Depends, HTTPException, Response, Query from typing import Optional, List import logging import pydantic from ..db_engine import db, SbaPlayer, model_to_dict, fn, chunked, query_to_csv from ..dependencies import ( oauth2_scheme, valid_token, PRIVATE_IN_SCHEMA, handle_db_errors, ) logger = logging.getLogger("discord_app") router = APIRouter(prefix="/api/v3/sbaplayers", tags=["sbaplayers"]) class SbaPlayerModel(pydantic.BaseModel): first_name: str last_name: str key_mlbam: Optional[int] = None key_fangraphs: Optional[int] = None key_bbref: Optional[str] = None key_retro: Optional[str] = None class PlayerList(pydantic.BaseModel): players: List[SbaPlayerModel] @router.get("") @handle_db_errors async def get_players( full_name: list = Query(default=None), first_name: list = Query(default=None), last_name: list = Query(default=None), key_fangraphs: list = Query(default=None), key_bbref: list = Query(default=None), key_retro: list = Query(default=None), key_mlbam: list = Query(default=None), sort: Optional[str] = None, csv: Optional[bool] = False, ): all_players = SbaPlayer.select() if full_name is not None: name_list = [x.lower() for x in full_name] all_players = all_players.where( fn.lower(SbaPlayer.first_name) + " " + fn.lower(SbaPlayer.last_name) << name_list ) if first_name is not None: name_list = [x.lower() for x in first_name] all_players = all_players.where(fn.lower(SbaPlayer.first_name) << name_list) if last_name is not None: name_list = [x.lower() for x in last_name] all_players = all_players.where(fn.lower(SbaPlayer.last_name) << name_list) if key_fangraphs is not None: all_players = all_players.where(SbaPlayer.key_fangraphs << key_fangraphs) if key_bbref is not None: name_list = [x.lower() for x in key_bbref] all_players = all_players.where(fn.lower(SbaPlayer.key_bbref) << name_list) if key_retro is not None: name_list = [x.lower() for x in key_retro] all_players = all_players.where(fn.lower(SbaPlayer.key_retro) << name_list) if key_mlbam is not None: all_players = all_players.where(SbaPlayer.key_mlbam << key_mlbam) if sort is not None: if sort in ["firstname-asc", "fullname-asc"]: all_players = all_players.order_by(SbaPlayer.first_name) elif sort in ["firstname-desc", "fullname-desc"]: all_players = all_players.order_by(-SbaPlayer.first_name) elif sort == "lastname-asc": all_players = all_players.order_by(SbaPlayer.last_name) elif sort == "lastname-desc": all_players = all_players.order_by(-SbaPlayer.last_name) elif sort == "fangraphs-asc": all_players = all_players.order_by(SbaPlayer.key_fangraphs) elif sort == "fangraphs-desc": all_players = all_players.order_by(-SbaPlayer.key_fangraphs) elif sort == "bbref-asc": all_players = all_players.order_by(SbaPlayer.key_bbref) elif sort == "bbref-desc": all_players = all_players.order_by(-SbaPlayer.key_bbref) elif sort == "retro-asc": all_players = all_players.order_by(SbaPlayer.key_retro) elif sort == "retro-desc": all_players = all_players.order_by(-SbaPlayer.key_retro) elif sort == "mlbam-asc": all_players = all_players.order_by(SbaPlayer.key_mlbam) elif sort == "mlbam-desc": all_players = all_players.order_by(-SbaPlayer.key_mlbam) if csv: return_val = query_to_csv(all_players) return Response(content=return_val, media_type="text/csv") return_val = { "count": all_players.count(), "players": [model_to_dict(x) for x in all_players], } return return_val @router.get("/{player_id}") @handle_db_errors async def get_one_player(player_id: int): this_player = SbaPlayer.get_or_none(SbaPlayer.id == player_id) if this_player is None: raise HTTPException( status_code=404, detail=f"SbaPlayer id {player_id} not found" ) r_data = model_to_dict(this_player) return r_data @router.patch("/{player_id}", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def patch_player( player_id: int, first_name: Optional[str] = None, last_name: Optional[str] = None, key_fangraphs: Optional[str] = None, key_bbref: Optional[str] = None, key_retro: Optional[str] = None, key_mlbam: Optional[str] = None, token: str = Depends(oauth2_scheme), ): if not valid_token(token): logging.warning(f"Bad Token: {token}") raise HTTPException( status_code=401, detail="You are not authorized to patch mlb players. This event has been logged.", ) this_player = SbaPlayer.get_or_none(SbaPlayer.id == player_id) if this_player is None: raise HTTPException( status_code=404, detail=f"SbaPlayer id {player_id} not found" ) if first_name is not None: this_player.first_name = first_name if last_name is not None: this_player.last_name = last_name if key_fangraphs is not None: this_player.key_fangraphs = key_fangraphs if key_bbref is not None: this_player.key_bbref = key_bbref if key_retro is not None: this_player.key_retro = key_retro if key_mlbam is not None: this_player.key_mlbam = key_mlbam if this_player.save() == 1: return_val = model_to_dict(this_player) return return_val else: raise HTTPException( status_code=418, detail="Well slap my ass and call me a teapot; I could not save that player", ) @router.post("/", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def post_players(players: PlayerList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"Bad Token: {token}") raise HTTPException( status_code=401, detail="You are not authorized to post mlb players. This event has been logged.", ) new_players = [] for x in players.players: dupes = SbaPlayer.select().where( ( (SbaPlayer.key_fangraphs == x.key_fangraphs) & (x.key_fangraphs is not None) ) | ((SbaPlayer.key_mlbam == x.key_mlbam) & (x.key_mlbam is not None)) | ((SbaPlayer.key_retro == x.key_retro) & (x.key_retro is not None)) | ((SbaPlayer.key_bbref == x.key_bbref) & (x.key_bbref is not None)) ) if dupes.count() > 0: logger.error(f"Found a dupe for {x}") raise HTTPException( status_code=400, detail=f"{x.first_name} {x.last_name} has a key already in the database", ) new_players.append(x.model_dump()) with db.atomic(): for batch in chunked(new_players, 15): SbaPlayer.insert_many(batch).on_conflict_ignore().execute() return f"Inserted {len(new_players)} new MLB players" @router.post("/one", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def post_one_player(player: SbaPlayerModel, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"Bad Token: {token}") raise HTTPException( status_code=401, detail="You are not authorized to post mlb players. This event has been logged.", ) dupes = SbaPlayer.select().where( (SbaPlayer.key_fangraphs == player.key_fangraphs) | (SbaPlayer.key_mlbam == player.key_mlbam) | (SbaPlayer.key_bbref == player.key_bbref) ) if dupes.count() > 0: logging.info(f"POST /SbaPlayers/one - dupes found:") for x in dupes: logging.info(f"{x}") raise HTTPException( status_code=400, detail=f"{player.first_name} {player.last_name} has a key already in the database", ) new_player = SbaPlayer(**player.dict()) saved = new_player.save() if saved == 1: return_val = model_to_dict(new_player) return return_val else: raise HTTPException( status_code=418, detail="Well slap my ass and call me a teapot; I could not save that player", ) @router.delete("/{player_id}", include_in_schema=PRIVATE_IN_SCHEMA) @handle_db_errors async def delete_player(player_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"Bad Token: {token}") raise HTTPException( status_code=401, detail="You are not authorized to delete mlb players. This event has been logged.", ) this_player = SbaPlayer.get_or_none(SbaPlayer.id == player_id) if this_player is None: raise HTTPException( status_code=404, detail=f"SbaPlayer id {player_id} not found" ) count = this_player.delete_instance() if count == 1: return f"Player {player_id} has been deleted" else: raise HTTPException( status_code=500, detail=f"Player {player_id} was not deleted" )