paper-dynasty-database/app/routers_v2/pitchingcards.py

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
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
    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), cardset_id: list = Query(default=None), short_output: bool = False):
    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)

    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 batting cards. This event has been logged.'
        )

    new_cards = []
    updates = 0

    for x in cards.cards:
        try:
            PitchingCard.get(
                (PitchingCard.player_id == x.player_id) & (PitchingCard.variant == x.variant)
            )
            updates += PitchingCard.update(x.dict()).where(
                (PitchingCard.player_id == x.player_id) & (PitchingCard.variant == x.variant)
            ).execute()
        except PitchingCard.DoesNotExist:
            new_cards.append(x.dict())

    with db.atomic():
        for batch in chunked(new_cards, 30):
            PitchingCard.insert_many(batch).on_conflict_replace().execute()

    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 batting 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'BattingCard 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')