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

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

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


class PlayerModel(pydantic.BaseModel):
    first_name: str
    last_name: str
    key_fangraphs: int = None
    key_bbref: str = None
    key_retro: str = None
    key_mlbam: int = None
    offense_col: int = None


class PlayerList(pydantic.BaseModel):
    players: List[PlayerModel]


@router.get('')
async def get_players(
        full_name: list = Query(default=None), first_name: list = Query(default=None),
        last_name: list = Query(default=None), key_fangraphs: list = Query(default=None),
        key_bbref: list = Query(default=None), key_retro: list = Query(default=None),
        key_mlbam: list = Query(default=None), offense_col: list = Query(default=None), csv: Optional[bool] = False):
    all_players = MlbPlayer.select()

    if full_name is not None:
        name_list = [x.lower() for x in full_name]
        all_players = all_players.where(
            fn.lower(MlbPlayer.first_name) + ' ' + fn.lower(MlbPlayer.last_name) << name_list
        )
    if first_name is not None:
        all_players = all_players.where(MlbPlayer.first_name << first_name)
    if first_name is not None:
        all_players = all_players.where(MlbPlayer.first_name << first_name)
    if last_name is not None:
        all_players = all_players.where(MlbPlayer.last_name << last_name)
    if key_fangraphs is not None:
        all_players = all_players.where(MlbPlayer.key_fangraphs << key_fangraphs)
    if key_bbref is not None:
        all_players = all_players.where(MlbPlayer.key_bbref << key_bbref)
    if key_retro is not None:
        all_players = all_players.where(MlbPlayer.key_retro << key_retro)
    if key_mlbam is not None:
        all_players = all_players.where(MlbPlayer.key_mlbam << key_mlbam)
    if offense_col is not None:
        all_players = all_players.where(MlbPlayer.offense_col << offense_col)

    if csv:
        return_val = query_to_csv(all_players)
        db.close()
        return Response(content=return_val, media_type='text/csv')

    return_val = {'count': all_players.count(), 'players': [
        model_to_dict(x) for x in all_players
    ]}
    db.close()
    return return_val


@router.get('/{player_id}')
async def get_one_player(player_id: int):
    this_player = MlbPlayer.get_or_none(MlbPlayer.id == player_id)
    if this_player is None:
        db.close()
        raise HTTPException(status_code=404, detail=f'MlbPlayer id {player_id} not found')

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


@router.patch('/{player_id}')
async def patch_player(
        player_id: int, first_name: Optional[str] = None, last_name: Optional[str] = None,
        key_fangraphs: Optional[str] = None, key_bbref: Optional[str] = None, key_retro: Optional[str] = None,
        key_mlbam: Optional[str] = None, offense_col: Optional[str] = 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 mlb players. This event has been logged.'
        )

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

    if first_name is not None:
        this_player.first_name = first_name
    if last_name is not None:
        this_player.last_name = last_name
    if key_fangraphs is not None:
        this_player.key_fangraphs = key_fangraphs
    if key_bbref is not None:
        this_player.key_bbref = key_bbref
    if key_retro is not None:
        this_player.key_retro = key_retro
    if key_mlbam is not None:
        this_player.key_mlbam = key_mlbam
    if offense_col is not None:
        this_player.offense_col = offense_col

    if this_player.save() == 1:
        return_val = model_to_dict(this_player)
        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 player'
        )


@router.post('')
async def post_players(players: PlayerList, 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 mlb players. This event has been logged.'
        )

    new_players = []
    for x in players.players:
        dupes = MlbPlayer.select().where(
            (MlbPlayer.key_fangraphs == x.key_fangraphs) | (MlbPlayer.key_mlbam == x.key_mlbam) |
            (MlbPlayer.key_retro == x.key_retro) | (MlbPlayer.key_bbref == x.key_bbref)
        )
        if dupes.count() > 0:
            db.close()
            raise HTTPException(
                status_code=400,
                detail=f'{x.first_name} {x.last_name} has a key already in the database'
            )

        new_players.append(x.dict())

    with db.atomic():
        for batch in chunked(new_players, 15):
            MlbPlayer.insert_many(batch).on_conflict_replace().execute()
    db.close()

    return f'Inserted {len(new_players)} new MLB players'


@router.delete('/{player_id}')
async def delete_player(player_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 mlb players. This event has been logged.'
        )

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

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

    if count == 1:
        raise HTTPException(status_code=200, detail=f'Player {player_id} has been deleted')
    else:
        raise HTTPException(status_code=500, detail=f'Player {player_id} was not deleted')