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

from fastapi import APIRouter, Depends, HTTPException, Response, Query
from typing import Literal, Optional, List
import logging
import pydantic
from pandas import DataFrame

from ..db_engine import db, BattingCard, model_to_dict, fn, chunked
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/battingcard',
    tags=['battingcard']
)


class BattingCardModel(pydantic.BaseModel):
    player_id: int
    steal_low: int = 3
    steal_high: int = 20
    steal_auto: bool = False
    steal_jump: float = 0
    bunting: str = 'C'
    hit_and_run: str = 'C'
    running: int = 10
    offense_col: int
    hand: Literal['R', 'L', 'S']


class BattingCardList(pydantic.BaseModel):
    cards: List[BattingCardModel]


@router.get('')
async def get_batting_cards(player_id: list = Query(default=None), short_output: bool = False):
    all_cards = BattingCard.select()
    if player_id is not None:
        all_cards = all_cards.where(BattingCard.player_id << player_id)

    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 = BattingCard.get_or_none(BattingCard.id == card_id)
    if this_card is None:
        db.close()
        raise HTTPException(status_code=404, detail=f'BattingCard 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 = BattingCard.select().where(BattingCard.player_id == player_id).order_by(BattingCard.variant)
    if variant is not None:
        all_cards = all_cards.where(BattingCard.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: BattingCardList, 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 = [x.dict() for x in cards.cards]
    with db.atomic():
        for batch in chunked(new_cards, 30):
            BattingCard.insert_many(batch).on_conflict_replace().execute()

    db.close()
    return f'Inserted {len(new_cards)} batting cards'


@router.patch('/{card_id}')
async def patch_card(
        card_id: int, steal_low: Optional[int] = None, steal_high: Optional[int] = None,
        steal_auto: Optional[bool] = None, steal_jump: Optional[float] = None, bunting: Optional[str] = None,
        hit_and_run: Optional[str] = None, running: Optional[int] = None, offense_col: Optional[int] = None,
        hand: Literal['R', 'L', 'S'] = 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 batting cards. This event has been logged.'
        )

    this_card = BattingCard.get_or_none(BattingCard.id == card_id)
    if this_card is None:
        db.close()
        raise HTTPException(status_code=404, detail=f'BattingCard id {card_id} not found')

    if steal_low is not None:
        this_card.steal_low = steal_low
    if steal_high is not None:
        this_card.steal_high = steal_high
    if steal_auto is not None:
        this_card.steal_auto = steal_auto
    if steal_jump is not None:
        this_card.steal_jump = steal_jump
    if bunting is not None:
        this_card.bunting = bunting
    if hit_and_run is not None:
        this_card.hit_and_run = hit_and_run
    if running is not None:
        this_card.running = running
    if offense_col is not None:
        this_card.offense_col = offense_col
    if hand is not None:
        this_card.hand = hand

    if this_card.save() == 1:
        return_val = model_to_dict(this_card)
        db.close()
        return return_val
    else:
        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 patch batting cards. This event has been logged.'
        )

    this_card = BattingCard.get_or_none(BattingCard.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')