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,
    limit: int = 100,
):
    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}")
        )

    total_count = all_games.count() if not csv else 0
    all_games = all_games.limit(max(0, min(limit, 500)))

    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": total_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"
        )