import random from fastapi import APIRouter, Depends, HTTPException, Query from typing import Literal, Optional, List import logging import pydantic from ..db_engine import db, PitchingCard, model_to_dict, chunked, Player, fn, MlbPlayer from ..db_helpers import upsert_pitching_cards 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/pitchingcards", tags=["pitchingcards"]) class PitchingCardModel(pydantic.BaseModel): player_id: int variant: int = 0 balk: int = 0 wild_pitch: int = 0 hold: int = 0 starter_rating: int = 1 relief_rating: int = 0 closer_rating: int = None batting: str = "#1WR-C" offense_col: int = None hand: Literal["R", "L", "S"] = "R" class PitchingCardList(pydantic.BaseModel): cards: List[PitchingCardModel] @router.get("") async def get_pitching_cards( player_id: list = Query(default=None), player_name: list = Query(default=None), cardset_id: list = Query(default=None), short_output: bool = False, limit: Optional[int] = None, ): all_cards = PitchingCard.select() if player_id is not None: all_cards = all_cards.where(PitchingCard.player_id << player_id) if cardset_id is not None: all_players = Player.select().where(Player.cardset_id << cardset_id) all_cards = all_cards.where(PitchingCard.player << all_players) if player_name is not None: name_list = [x.lower() for x in player_name] all_players = Player.select().where(fn.lower(Player.p_name) << name_list) all_cards = all_cards.where(PitchingCard.player << all_players) if limit is not None: all_cards = all_cards.limit(limit) return_val = { "count": all_cards.count(), "cards": [model_to_dict(x, recurse=not short_output) for x in all_cards], } db.close() return return_val @router.get("/{card_id}") async def get_one_card(card_id: int): this_card = PitchingCard.get_or_none(PitchingCard.id == card_id) if this_card is None: db.close() raise HTTPException( status_code=404, detail=f"PitchingCard id {card_id} not found" ) r_card = model_to_dict(this_card) db.close() return r_card @router.get("/player/{player_id}") async def get_player_cards( player_id: int, variant: list = Query(default=None), short_output: bool = False ): all_cards = ( PitchingCard.select() .where(PitchingCard.player_id == player_id) .order_by(PitchingCard.variant) ) if variant is not None: all_cards = all_cards.where(PitchingCard.variant << variant) return_val = { "count": all_cards.count(), "cards": [model_to_dict(x, recurse=not short_output) for x in all_cards], } db.close() return return_val @router.put("") async def put_cards(cards: PitchingCardList, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"Bad Token: {token}") db.close() raise HTTPException( status_code=401, detail="You are not authorized to post pitching cards. This event has been logged.", ) new_cards = [] updates = 0 for x in cards.cards: try: old = PitchingCard.get( (PitchingCard.player_id == x.player_id) & (PitchingCard.variant == x.variant) ) if x.offense_col is None: x.offense_col = old.offense_col updates += ( PitchingCard.update(x.dict()) .where( (PitchingCard.player_id == x.player_id) & (PitchingCard.variant == x.variant) ) .execute() ) except PitchingCard.DoesNotExist: if x.offense_col is None: this_player = Player.get_or_none(Player.player_id == x.player_id) mlb_player = MlbPlayer.get_or_none( MlbPlayer.key_bbref == this_player.bbref_id ) if mlb_player is not None: logging.info( f"setting offense_col to {mlb_player.offense_col} for {this_player.p_name}" ) x.offense_col = mlb_player.offense_col else: logging.info( f"randomly setting offense_col for {this_player.p_name}" ) x.offense_col = random.randint(1, 3) logging.debug(f"x.dict(): {x.dict()}") new_cards.append(x.dict()) with db.atomic(): # Use PostgreSQL-compatible upsert helper upsert_pitching_cards(new_cards, batch_size=30) db.close() return f"Updated cards: {updates}; new cards: {len(new_cards)}" @router.patch("/{card_id}") async def patch_card( card_id: int, balk: Optional[int] = None, wild_pitch: Optional[int] = None, hold: Optional[int] = None, starter_rating: Optional[int] = None, relief_rating: Optional[int] = None, closer_rating: Optional[int] = None, batting: Optional[int] = None, token: str = Depends(oauth2_scheme), ): if not valid_token(token): logging.warning(f"Bad Token: {token}") db.close() raise HTTPException( status_code=401, detail="You are not authorized to patch pitching cards. This event has been logged.", ) this_card = PitchingCard.get_or_none(PitchingCard.id == card_id) if this_card is None: db.close() raise HTTPException( status_code=404, detail=f"PitchingCard id {card_id} not found" ) if balk is not None: this_card.balk = balk if wild_pitch is not None: this_card.wild_pitch = wild_pitch if hold is not None: this_card.hold = hold if starter_rating is not None: this_card.starter_rating = starter_rating if relief_rating is not None: this_card.relief_rating = relief_rating if closer_rating is not None: this_card.closer_rating = closer_rating if batting is not None: this_card.batting = batting if this_card.save() == 1: return_val = model_to_dict(this_card) db.close() return return_val else: db.close() raise HTTPException( status_code=418, detail="Well slap my ass and call me a teapot; I could not save that card", ) @router.delete("/{card_id}") async def delete_card(card_id: int, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"Bad Token: {token}") db.close() raise HTTPException( status_code=401, detail="You are not authorized to delete pitching cards. This event has been logged.", ) this_card = PitchingCard.get_or_none(PitchingCard.id == card_id) if this_card is None: db.close() raise HTTPException(status_code=404, detail=f"Pitching id {card_id} not found") count = this_card.delete_instance() db.close() if count == 1: return f"Card {this_card} has been deleted" else: raise HTTPException( status_code=500, detail=f"Card {this_card} could not be deleted" ) @router.delete("") async def delete_all_cards(token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f"Bad Token: {token}") db.close() raise HTTPException( status_code=401, detail="You are not authorized to delete pitching cards. This event has been logged.", ) d_query = PitchingCard.delete() d_query.execute() return f"Deleted {d_query.count()} pitching cards"