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

from fastapi import APIRouter, Depends, HTTPException, Query
from typing import Literal, Optional, List
import logging
import pydantic
from pydantic import validator, root_validator

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


class BattingCardRatingsModel(pydantic.BaseModel):
    battingcard_id: int
    vs_hand: Literal['R', 'L', 'vR', 'vL']
    homerun: float = 0.0
    bp_homerun: float = 0.0
    triple: float = 0.0
    double_three: float = 0.0
    double_two: float = 0.0
    double_pull: float = 0.0
    single_two: float = 0.0
    single_one: float = 0.0
    single_center: float = 0.0
    bp_single: float = 0.0
    hbp: float = 0.0
    walk: float = 0.0
    strikeout: float = 0.0
    lineout: float = 0.0
    popout: float = 0.0
    flyout_a: float = 0.0
    flyout_bq: float = 0.0
    flyout_lf_b: float = 0.0
    flyout_rf_b: float = 0.0
    groundout_a: float = 0.0
    groundout_b: float = 0.0
    groundout_c: float = 0.0
    avg: float = 0.0
    obp: float = 0.0
    slg: float = 0.0

    @validator("avg", always=True)
    def avg_validator(cls, v, values, **kwargs):
        return (values['homerun'] + values['bp_homerun'] / 2 + values['triple'] + values['double_three'] +
                values['double_two'] + values['double_pull'] + values['single_two'] + values['single_one'] +
                values['single_center'] + values['bp_single'] / 2) / 108

    @validator("obp", always=True)
    def obp_validator(cls, v, values, **kwargs):
        return ((values['hbp'] + values['walk']) / 108) + values['avg']

    @validator("slg", always=True)
    def slg_validator(cls, v, values, **kwargs):
        return (values['homerun'] * 4 + values['bp_homerun'] * 2 + values['triple'] * 3 + values['double_three'] * 2 +
                values['double_two'] * 2 + values['double_pull'] * 2 + values['single_two'] + values['single_one'] +
                values['single_center'] + values['bp_single'] / 2) / 108

    @root_validator
    def validate_chance_total(cls, values):
        total_chances = (
                values['homerun'] + values['bp_homerun'] + values['triple'] + values['double_three'] +
                values['double_two'] + values['double_pull'] + values['single_two'] + values['single_one'] +
                values['single_center'] + values['bp_single'] + values['hbp'] + values['walk'] +
                values['strikeout'] + values['lineout'] + values['popout'] + values['flyout_a'] +
                values['flyout_bq'] + values['flyout_lf_b'] + values['flyout_rf_b'] + values['groundout_a'] +
                values['groundout_b'] + values['groundout_c'])

        if total_chances != 108:
            raise ValueError("Must have exactly 108 chances on the card")
        return values


class RatingsList(pydantic.BaseModel):
    ratings: List[BattingCardRatingsModel]


@router.get('')
async def get_card_ratings(
        battingcard_id: list = Query(default=None), vs_hand: Literal['R', 'L', 'vR', 'vL'] = None,
        short_output: bool = False, 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 pull card ratings.'
        )

    all_ratings = BattingCardRatings.select()

    if battingcard_id is not None:
        all_ratings = all_ratings.where(BattingCardRatings.battingcard_id << battingcard_id)
    if vs_hand is not None:
        all_ratings = all_ratings.where(BattingCardRatings.vs_hand << vs_hand[-1])

    return_val = {'count': all_ratings.count(), 'ratings': [
        model_to_dict(x, recurse=not short_output) for x in all_ratings
    ]}
    db.close()
    return return_val


@router.get('/{ratings_id}')
async def get_one_rating(ratings_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 pull card ratings.'
        )

    this_rating = BattingCardRatings.get_or_none(BattingCardRatings.id == ratings_id)
    if this_rating is None:
        db.close()
        raise HTTPException(status_code=404, detail=f'BattingCardRating id {ratings_id} not found')

    r_data = model_to_dict(this_rating)
    db.close()
    return r_data


@router.get('/{player_id}')
async def get_player_ratings(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)

    all_ratings = BattingCardRatings.select().where(BattingCardRatings.battingcard << all_cards)

    return_val = {'count': all_ratings.count(), 'ratings': [
        model_to_dict(x, recurse=not short_output) for x in all_ratings
    ]}
    db.close()
    return return_val


@router.put('')
async def put_ratings(ratings: RatingsList, 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 card ratings.'
        )

    new_ratings = [x.dict() for x in ratings.ratings]
    with db.atomic():
        for batch in chunked(new_ratings, 30):
            BattingCardRatings.insert_many(batch).on_conflict_replace().execute()

    db.close()
    return f'Inserted {len(new_ratings)} batting ratings'


@router.delete('/{ratings_id}')
async def put_one_rating(
        ratings_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 post card ratings.'
        )

    this_rating = BattingCardRatings.get_or_none(BattingCardRatings.id == ratings_id)
    if this_rating is None:
        db.close()
        raise HTTPException(status_code=404, detail=f'BattingCardRating id {ratings_id} not found')

    count = this_rating.delete_instance()
    db.close()

    if count == 1:
        return f'Rating {this_rating} has been deleted'
    else:
        raise HTTPException(status_code=500, detail=f'Rating {this_rating} could not be deleted')