193 lines
6.0 KiB
Python
193 lines
6.0 KiB
Python
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}")
|
|
)
|
|
|
|
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": 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"
|
|
)
|