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

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

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


class CardPositionModel(pydantic.BaseModel):
    player_id: int
    variant: int = 0
    position: Literal['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF', 'DH']
    innings: int = 1
    range: int = 5
    error: int = 0
    arm: Optional[int] = None
    pb: Optional[int] = None
    overthrow: Optional[int] = None

    @root_validator
    def position_validator(cls, values):
        if values['position'] in ['C', 'LF', 'CF', 'RF'] and values['arm'] is None:
            raise ValueError(f'{values["position"]} must have an arm rating')
        if values['position'] == 'C' and (values['pb'] is None or values['overthrow'] is None):
            raise ValueError('Catchers must have a pb and overthrow rating')
        return values


class PositionList(pydantic.BaseModel):
    positions: List[CardPositionModel]


@router.get('')
async def get_card_positions(
        player_id: list = Query(default=None), position: list = Query(default=None), min_innings: Optional[int] = 1,
        r: list = Query(default=None), e: list = Query(default=None), arm: list = Query(default=None),
        pb: list = Query(default=None), overthrow: list = Query(default=None), cardset_id: list = Query(default=None),
        short_output: Optional[bool] = False, sort: Optional[str] = 'innings-desc'):
    all_pos = CardPosition.select().where(CardPosition.innings >= min_innings).order_by(
        CardPosition.player, CardPosition.position, CardPosition.variant
    )

    if player_id is not None:
        all_pos = all_pos.where(CardPosition.player_id << player_id)
    if position is not None:
        p_list = [x.lower() for x in position]
        all_pos = all_pos.where(fn.Lower(CardPosition.position) << p_list)
    if r is not None:
        all_pos = all_pos.where(CardPosition.range << r)
    if e is not None:
        all_pos = all_pos.where(CardPosition.error << e)
    if arm is not None:
        all_pos = all_pos.where(CardPosition.arm << arm)
    if pb is not None:
        all_pos = all_pos.where(CardPosition.pb << pb)
    if overthrow is not None:
        all_pos = all_pos.where(CardPosition.overthrow << overthrow)
    if cardset_id is not None:
        all_players = Player.select().where(Player.cardset_id << cardset_id)
        all_pos = all_pos.where(CardPosition.player << all_players)

    if sort == 'innings-desc':
        all_pos = all_pos.order_by(CardPosition.innings.desc())
    elif sort == 'innings-asc':
        all_pos = all_pos.order_by(CardPosition.innings)
    elif sort == 'range-desc':
        all_pos = all_pos.order_by(CardPosition.range.desc())
    elif sort == 'range-asc':
        all_pos = all_pos.order_by(CardPosition.range)

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


@router.get('/{position_id}')
async def get_one_position(position_id: int):
    this_pos = CardPosition.get_or_none(CardPosition.id == position_id)
    if this_pos is None:
        db.close()
        raise HTTPException(status_code=404, detail=f'CardPosition id {position_id} not found')

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


@router.put('')
async def put_positions(positions: PositionList, 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 positions. This event has been logged.'
        )

    new_cards = []
    updates = 0

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

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

    db.close()
    return f'Updated cards: {updates}; new cards: {len(new_cards)}'


@router.delete('/{position_id}')
async def delete_position(position_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 card positions. This event has been logged.'
        )

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

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

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