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

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

from ..db_engine import StratGame, model_to_dict, fn
from ..dependencies import oauth2_scheme, valid_token


router = APIRouter(
    prefix='/api/v2/games',
    tags=['games']
)


class GameModel(pydantic.BaseModel):
    season: int
    game_type: str
    away_team_id: int
    home_team_id: int
    week: int = 1
    away_score: int = 0
    home_score: int = 0
    away_team_value: int = None
    home_team_value: int = None
    away_team_ranking: int = None
    home_team_ranking: int = None
    ranked: bool = False
    short_game: bool = False
    forfeit: bool = False


class GameList(pydantic.BaseModel):
    games: List[GameModel]


@router.get('')
async def get_games(
        season: list = Query(default=None), forfeit: Optional[bool] = None, away_team_id: list = Query(default=None),
        home_team_id: list = Query(default=None), team1_id: list = Query(default=None),
        team2_id: list = Query(default=None), game_type: list = Query(default=None), ranked: Optional[bool] = None,
        short_game: Optional[bool] = None, csv: Optional[bool] = False, short_output: bool = False,
        gauntlet_id: Optional[int] = None):
    all_games = StratGame.select().order_by(StratGame.id)

    if season is not None:
        all_games = all_games.where(StratGame.season << season)
    if forfeit is not None:
        all_games = all_games.where(StratGame.forfeit == forfeit)
    if away_team_id is not None:
        all_games = all_games.where(StratGame.away_team_id << away_team_id)
    if home_team_id is not None:
        all_games = all_games.where(StratGame.home_team_id << home_team_id)
    if team1_id is not None:
        all_games = all_games.where(
            (StratGame.away_team_id << team1_id) | (StratGame.home_team_id << team1_id)
        )
    if team2_id is not None:
        all_games = all_games.where(
            (StratGame.away_team_id << team2_id) | (StratGame.home_team_id << team2_id)
        )
    if game_type is not None:
        g_list = [x.lower() for x in game_type]
        all_games = all_games.where(fn.Lower(StratGame.game_type) << g_list)
    if ranked is not None:
        all_games = all_games.where(StratGame.ranked == ranked)
    if short_game is not None:
        all_games = all_games.where(StratGame.short_game == short_game)
    if gauntlet_id is not None:
        all_games = all_games.where(StratGame.game_type.contains(f'gauntlet-{gauntlet_id}'))

    if csv:
        return_vals = [model_to_dict(x) for x in all_games]
        for x in return_vals:
            x['away_abbrev'] = x['away_team']['abbrev']
            x['home_abbrev'] = x['home_team']['abbrev']
            del x['away_team'], x['home_team']

        output = pd.DataFrame(return_vals)[[
            'id', 'away_abbrev', 'home_abbrev', 'away_score', 'home_score', 'away_team_value', 'home_team_value',
            'game_type', 'season', 'week', 'short_game', 'ranked'
        ]]

        return Response(content=output.to_csv(index=False), media_type='text/csv')

    return_val = {'count': all_games.count(), 'games': [
        model_to_dict(x, recurse=not short_output) for x in all_games
    ]}
    return return_val


@router.get('/{game_id}')
async def get_one_game(game_id: int):
    this_game = StratGame.get_or_none(StratGame.id == game_id)
    if not this_game:
        raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found')

    g_result = model_to_dict(this_game)
    return g_result


@router.patch('/{game_id}')
async def patch_game(
        game_id: int, game_type: Optional[str] = None, away_score: Optional[int] = None,
        home_score: Optional[int] = None, token: str = Depends(oauth2_scheme)):
    if not valid_token(token):
        logging.warning('patch_game - Bad Token: [REDACTED]')
        raise HTTPException(status_code=401, detail='Unauthorized')

    this_game = StratGame.get_or_none(StratGame.id == game_id)
    if not this_game:
        raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found')

    if away_score is not None:
        this_game.away_score = away_score
    if home_score is not None:
        this_game.home_score = home_score
    if game_type is not None:
        this_game.game_type = game_type

    if this_game.save() == 1:
        g_result = model_to_dict(this_game)
        return g_result
    else:
        raise HTTPException(status_code=500, detail=f'Unable to patch game {game_id}')


@router.post('')
async def post_game(this_game: GameModel, token: str = Depends(oauth2_scheme)):
    if not valid_token(token):
        logging.warning('post_games - Bad Token: [REDACTED]')
        raise HTTPException(status_code=401, detail='Unauthorized')

    this_game = StratGame(**this_game.dict())

    saved = this_game.save()
    if saved == 1:
        return_val = model_to_dict(this_game)
        return return_val
    else:
        raise HTTPException(
            status_code=418,
            detail='Well slap my ass and call me a teapot; I could not save that game'
        )


@router.delete('/{game_id}')
async def delete_game(game_id: int, token: str = Depends(oauth2_scheme)):
    if not valid_token(token):
        logging.warning('delete_game - Bad Token: [REDACTED]')
        raise HTTPException(status_code=401, detail='Unauthorized')

    this_game = StratGame.get_or_none(StratGame.id == game_id)
    if not this_game:
        raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found')

    count = this_game.delete_instance()

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