dcaf184ad3
All checks were successful
Build Docker Image / build (pull_request) Successful in 2m11s
Closes #73 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from typing import List, Optional
|
|
import logging
|
|
|
|
from ..db_engine import db, Standings, Team, Division, model_to_dict, chunked, fn
|
|
from ..dependencies import (
|
|
oauth2_scheme,
|
|
valid_token,
|
|
PRIVATE_IN_SCHEMA,
|
|
handle_db_errors,
|
|
)
|
|
|
|
logger = logging.getLogger("discord_app")
|
|
|
|
router = APIRouter(prefix="/api/v3/standings", tags=["standings"])
|
|
|
|
|
|
@router.get("")
|
|
@handle_db_errors
|
|
async def get_standings(
|
|
season: int,
|
|
team_id: list = Query(default=None),
|
|
league_abbrev: Optional[str] = None,
|
|
division_abbrev: Optional[str] = None,
|
|
short_output: Optional[bool] = False,
|
|
):
|
|
standings = Standings.select_season(season)
|
|
|
|
# if standings.count() == 0:
|
|
# db.close()
|
|
# raise HTTPException(status_code=404, detail=f'No output for season {season}')
|
|
|
|
if team_id is not None:
|
|
t_query = Team.select().where(Team.id << team_id)
|
|
standings = standings.where(Standings.team << t_query)
|
|
|
|
if league_abbrev is not None:
|
|
l_query = Division.select().where(
|
|
fn.Lower(Division.league_abbrev) == league_abbrev.lower()
|
|
)
|
|
standings = standings.where(Standings.team.division << l_query)
|
|
|
|
if division_abbrev is not None:
|
|
d_query = Division.select().where(
|
|
fn.Lower(Division.division_abbrev) == division_abbrev.lower()
|
|
)
|
|
standings = standings.where(Standings.team.division << d_query)
|
|
|
|
def win_pct(this_team_stan):
|
|
if this_team_stan.wins + this_team_stan.losses == 0:
|
|
return 0
|
|
else:
|
|
return (
|
|
this_team_stan.wins / (this_team_stan.wins + this_team_stan.losses)
|
|
) + (this_team_stan.run_diff * 0.000001)
|
|
|
|
div_teams = [x for x in standings]
|
|
div_teams.sort(key=lambda team: win_pct(team), reverse=True)
|
|
|
|
return_standings = {
|
|
"count": len(div_teams),
|
|
"standings": [model_to_dict(x, recurse=not short_output) for x in div_teams],
|
|
}
|
|
|
|
db.close()
|
|
return return_standings
|
|
|
|
|
|
@router.get("/team/{team_id}")
|
|
@handle_db_errors
|
|
async def get_team_standings(team_id: int):
|
|
this_stan = Standings.get_or_none(Standings.team_id == team_id)
|
|
if this_stan is None:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No standings found for team id {team_id}"
|
|
)
|
|
|
|
return model_to_dict(this_stan)
|
|
|
|
|
|
@router.patch("/{stan_id}", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def patch_standings(
|
|
stan_id: int,
|
|
wins: Optional[int] = None,
|
|
losses: Optional[int] = None,
|
|
token: str = Depends(oauth2_scheme),
|
|
):
|
|
if not valid_token(token):
|
|
logger.warning(f"patch_standings - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
try:
|
|
this_stan = Standings.get_by_id(stan_id)
|
|
except Exception as e:
|
|
db.close()
|
|
raise HTTPException(status_code=404, detail=f"No team found with id {stan_id}")
|
|
|
|
if wins:
|
|
this_stan.wins = wins
|
|
if losses:
|
|
this_stan.losses = losses
|
|
|
|
this_stan.save()
|
|
db.close()
|
|
|
|
return model_to_dict(this_stan)
|
|
|
|
|
|
@router.post("/s{season}/new", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def post_standings(season: int, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logger.warning(f"post_standings - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
new_teams = []
|
|
all_teams = Team.select().where(Team.season == season)
|
|
for x in all_teams:
|
|
new_teams.append(Standings({"team_id": x.id}))
|
|
|
|
with db.atomic():
|
|
for batch in chunked(new_teams, 16):
|
|
Standings.insert_many(batch).on_conflict_ignore().execute()
|
|
db.close()
|
|
|
|
return f"Inserted {len(new_teams)} standings"
|
|
|
|
|
|
@router.post("/s{season}/recalculate", include_in_schema=PRIVATE_IN_SCHEMA)
|
|
@handle_db_errors
|
|
async def recalculate_standings(season: int, token: str = Depends(oauth2_scheme)):
|
|
if not valid_token(token):
|
|
logger.warning(f"recalculate_standings - Bad Token: {token}")
|
|
raise HTTPException(status_code=401, detail="Unauthorized")
|
|
|
|
code = Standings.recalculate(season)
|
|
db.close()
|
|
if code == 69:
|
|
raise HTTPException(status_code=500, detail=f"Error recreating Standings rows")
|
|
return f"Just recalculated standings for season {season}"
|