cal/paper-dynasty-database
1
0
Fork 0
Code Issues 7 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat: add limit/pagination to results endpoint (#137)

Browse Source 
Closes #137

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-03-24 05:02:35 -05:00
parent 6a217f97ee
commit 15aac6cb73
1 changed files with 198 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

283
app/routers_v2/results.py
View File

@ -8,10 +8,7 @@ from ..db_engine import Result, model_to_dict, Team, DataError, DoesNotExist
from ..dependencies import oauth2_scheme, valid_token
router = APIRouter(
prefix='/api/v2/results',
tags=['results']
)
router = APIRouter(prefix="/api/v2/results", tags=["results"])
class ResultModel(pydantic.BaseModel):
@ -31,15 +28,29 @@ class ResultModel(pydantic.BaseModel):
game_type: str
@router.get('')
@router.get("")
async def get_results(
away_team_id: Optional[int] = None, home_team_id: Optional[int] = None, team_one_id: Optional[int] = None,
team_two_id: Optional[int] = None, away_score_min: Optional[int] = None, away_score_max: Optional[int] = None,
home_score_min: Optional[int] = None, home_score_max: Optional[int] = None, bothscore_min: Optional[int] = None,
bothscore_max: Optional[int] = None, season: Optional[int] = None, week: Optional[int] = None,
week_start: Optional[int] = None, week_end: Optional[int] = None, ranked: Optional[bool] = None,
short_game: Optional[bool] = None, game_type: Optional[str] = None, vs_ai: Optional[bool] = None,
csv: Optional[bool] = None):
away_team_id: Optional[int] = None,
home_team_id: Optional[int] = None,
team_one_id: Optional[int] = None,
team_two_id: Optional[int] = None,
away_score_min: Optional[int] = None,
away_score_max: Optional[int] = None,
home_score_min: Optional[int] = None,
home_score_max: Optional[int] = None,
bothscore_min: Optional[int] = None,
bothscore_max: Optional[int] = None,
season: Optional[int] = None,
week: Optional[int] = None,
week_start: Optional[int] = None,
week_end: Optional[int] = None,
ranked: Optional[bool] = None,
short_game: Optional[bool] = None,
game_type: Optional[str] = None,
vs_ai: Optional[bool] = None,
csv: Optional[bool] = None,
limit: int = 100,
):
all_results = Result.select()
# if all_results.count() == 0:
@ -51,28 +62,40 @@ async def get_results(
this_team = Team.get_by_id(away_team_id)
all_results = all_results.where(Result.away_team == this_team)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No team found with id {away_team_id}')
raise HTTPException(
status_code=404, detail=f"No team found with id {away_team_id}"
)
if home_team_id is not None:
try:
this_team = Team.get_by_id(home_team_id)
all_results = all_results.where(Result.home_team == this_team)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No team found with id {home_team_id}')
raise HTTPException(
status_code=404, detail=f"No team found with id {home_team_id}"
)
if team_one_id is not None:
try:
this_team = Team.get_by_id(team_one_id)
all_results = all_results.where((Result.home_team == this_team) | (Result.away_team == this_team))
all_results = all_results.where(
(Result.home_team == this_team) | (Result.away_team == this_team)
)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No team found with id {team_one_id}')
raise HTTPException(
status_code=404, detail=f"No team found with id {team_one_id}"
)
if team_two_id is not None:
try:
this_team = Team.get_by_id(team_two_id)
all_results = all_results.where((Result.home_team == this_team) | (Result.away_team == this_team))
all_results = all_results.where(
(Result.home_team == this_team) | (Result.away_team == this_team)
)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No team found with id {team_two_id}')
raise HTTPException(
status_code=404, detail=f"No team found with id {team_two_id}"
)
if away_score_min is not None:
all_results = all_results.where(Result.away_score >= away_score_min)
@ -87,10 +110,14 @@ async def get_results(
all_results = all_results.where(Result.home_score <= home_score_max)
if bothscore_min is not None:
all_results = all_results.where((Result.home_score >= bothscore_min) & (Result.away_score >= bothscore_min))
all_results = all_results.where(
(Result.home_score >= bothscore_min) & (Result.away_score >= bothscore_min)
)
if bothscore_max is not None:
all_results = all_results.where((Result.home_score <= bothscore_max) & (Result.away_score <= bothscore_max))
all_results = all_results.where(
(Result.home_score <= bothscore_max) & (Result.away_score <= bothscore_max)
)
if season is not None:
all_results = all_results.where(Result.season == season)
@ -114,6 +141,8 @@ async def get_results(
all_results = all_results.where(Result.game_type == game_type)
all_results = all_results.order_by(Result.id)
limit = max(0, min(limit, 500))
all_results = all_results.limit(limit)
# Not functional
# if vs_ai is not None:
# AwayTeam = Team.alias()
@ -134,60 +163,115 @@ async def get_results(
# logging.info(f'Result Query:\n\n{all_results}')
if csv:
data_list = [['id', 'away_abbrev', 'home_abbrev', 'away_score', 'home_score', 'away_tv', 'home_tv',
'game_type', 'season', 'week', 'short_game', 'ranked']]
data_list = [
[
"id",
"away_abbrev",
"home_abbrev",
"away_score",
"home_score",
"away_tv",
"home_tv",
"game_type",
"season",
"week",
"short_game",
"ranked",
]
]
for line in all_results:
data_list.append([
line.id, line.away_team.abbrev, line.home_team.abbrev, line.away_score, line.home_score,
line.away_team_value, line.home_team_value, line.game_type if line.game_type else 'minor-league',
line.season, line.week, line.short_game, line.ranked
])
data_list.append(
[
line.id,
line.away_team.abbrev,
line.home_team.abbrev,
line.away_score,
line.home_score,
line.away_team_value,
line.home_team_value,
line.game_type if line.game_type else "minor-league",
line.season,
line.week,
line.short_game,
line.ranked,
]
)
return_val = DataFrame(data_list).to_csv(header=False, index=False)
return Response(content=return_val, media_type='text/csv')
return Response(content=return_val, media_type="text/csv")
else:
return_val = {'count': all_results.count(), 'results': []}
return_val = {"count": all_results.count(), "results": []}
for x in all_results:
return_val['results'].append(model_to_dict(x))
return_val["results"].append(model_to_dict(x))
return return_val
@router.get('/{result_id}')
@router.get("/{result_id}")
async def get_one_results(result_id, csv: Optional[bool] = None):
try:
this_result = Result.get_by_id(result_id)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No result found with id {result_id}')
raise HTTPException(
status_code=404, detail=f"No result found with id {result_id}"
)
if csv:
data_list = [
['id', 'away_abbrev', 'home_abbrev', 'away_score', 'home_score', 'away_tv', 'home_tv', 'game_type',
'season', 'week', 'game_type'],
[this_result.id, this_result.away_team.abbrev, this_result.away_team.abbrev, this_result.away_score,
this_result.home_score, this_result.away_team_value, this_result.home_team_value,
this_result.game_type if this_result.game_type else 'minor-league',
this_result.season, this_result.week, this_result.game_type]
[
"id",
"away_abbrev",
"home_abbrev",
"away_score",
"home_score",
"away_tv",
"home_tv",
"game_type",
"season",
"week",
"game_type",
],
[
this_result.id,
this_result.away_team.abbrev,
this_result.away_team.abbrev,
this_result.away_score,
this_result.home_score,
this_result.away_team_value,
this_result.home_team_value,
this_result.game_type if this_result.game_type else "minor-league",
this_result.season,
this_result.week,
this_result.game_type,
],
]
return_val = DataFrame(data_list).to_csv(header=False, index=False)
return Response(content=return_val, media_type='text/csv')
return Response(content=return_val, media_type="text/csv")
else:
return_val = model_to_dict(this_result)
return return_val
@router.get('/team/{team_id}')
@router.get("/team/{team_id}")
async def get_team_results(
team_id: int, season: Optional[int] = None, week: Optional[int] = None, csv: Optional[bool] = False):
all_results = Result.select().where((Result.away_team_id == team_id) | (Result.home_team_id == team_id)).order_by(Result.id)
team_id: int,
season: Optional[int] = None,
week: Optional[int] = None,
csv: Optional[bool] = False,
):
all_results = (
Result.select()
.where((Result.away_team_id == team_id) | (Result.home_team_id == team_id))
.order_by(Result.id)
)
try:
this_team = Team.get_by_id(team_id)
except DoesNotExist as e:
logging.error(f'Unknown team id {team_id} trying to pull team results')
raise HTTPException(404, f'Team id {team_id} not found')
except DoesNotExist:
logging.error(f"Unknown team id {team_id} trying to pull team results")
raise HTTPException(404, f"Team id {team_id} not found")
if season is not None:
all_results = all_results.where(Result.season == season)
@ -224,31 +308,38 @@ async def get_team_results(
if csv:
data_list = [
['team_id', 'ranked_wins', 'ranked_losses', 'casual_wins', 'casual_losses', 'team_ranking'],
[team_id, r_wins, r_loss, c_wins, c_loss, this_team.ranking]
[
"team_id",
"ranked_wins",
"ranked_losses",
"casual_wins",
"casual_losses",
"team_ranking",
],
[team_id, r_wins, r_loss, c_wins, c_loss, this_team.ranking],
]
return_val = DataFrame(data_list).to_csv(header=False, index=False)
return Response(content=return_val, media_type='text/csv')
return Response(content=return_val, media_type="text/csv")
else:
return_val = {
'team': model_to_dict(this_team),
'ranked_wins': r_wins,
'ranked_losses': r_loss,
'casual_wins': c_wins,
'casual_losses': c_loss,
"team": model_to_dict(this_team),
"ranked_wins": r_wins,
"ranked_losses": r_loss,
"casual_wins": c_wins,
"casual_losses": c_loss,
}
return return_val
@router.post('')
@router.post("")
async def post_result(result: ResultModel, token: str = Depends(oauth2_scheme)):
if not valid_token(token):
logging.warning('Bad Token: [REDACTED]')
logging.warning("Bad Token: [REDACTED]")
raise HTTPException(
status_code=401,
detail='You are not authorized to post results. This event has been logged.'
detail="You are not authorized to post results. This event has been logged.",
)
this_result = Result(**result.__dict__)
@ -256,24 +347,28 @@ async def post_result(result: ResultModel, token: str = Depends(oauth2_scheme)):
if result.ranked:
if not result.away_team_ranking:
error = f'Ranked game did not include away team ({result.away_team_id}) ranking.'
error = f"Ranked game did not include away team ({result.away_team_id}) ranking."
logging.error(error)
raise DataError(error)
if not result.home_team_ranking:
error = f'Ranked game did not include home team ({result.home_team_id}) ranking.'
error = f"Ranked game did not include home team ({result.home_team_id}) ranking."
logging.error(error)
raise DataError(error)
k_value = 20 if result.short_game else 60
ratio = (result.home_team_ranking - result.away_team_ranking) / 400
exp_score = 1 / (1 + (10 ** ratio))
exp_score = 1 / (1 + (10**ratio))
away_win = True if result.away_score > result.home_score else False
total_delta = k_value * exp_score
high_delta = total_delta * exp_score if exp_score > .5 else total_delta * (1 - exp_score)
high_delta = (
total_delta * exp_score
if exp_score > 0.5
else total_delta * (1 - exp_score)
)
low_delta = total_delta - high_delta
# exp_score > .5 means away team is favorite
if exp_score > .5 and away_win:
if exp_score > 0.5 and away_win:
final_delta = low_delta
away_delta = low_delta * 3
home_delta = -low_delta
@ -281,7 +376,7 @@ async def post_result(result: ResultModel, token: str = Depends(oauth2_scheme)):
final_delta = high_delta
away_delta = high_delta * 3
home_delta = -high_delta
elif exp_score <= .5 and not away_win:
elif exp_score <= 0.5 and not away_win:
final_delta = low_delta
away_delta = -low_delta
home_delta = low_delta * 3
@ -294,18 +389,20 @@ async def post_result(result: ResultModel, token: str = Depends(oauth2_scheme)):
away_delta = 0
home_delta = 0
logging.debug(f'/results ranking deltas\n\nk_value: {k_value} / ratio: {ratio} / '
f'exp_score: {exp_score} / away_win: {away_win} / total_delta: {total_delta} / '
f'high_delta: {high_delta} / low_delta: {low_delta} / final_delta: {final_delta} / ')
logging.debug(
f"/results ranking deltas\n\nk_value: {k_value} / ratio: {ratio} / "
f"exp_score: {exp_score} / away_win: {away_win} / total_delta: {total_delta} / "
f"high_delta: {high_delta} / low_delta: {low_delta} / final_delta: {final_delta} / "
)
away_team = Team.get_by_id(result.away_team_id)
away_team.ranking += away_delta
away_team.save()
logging.info(f'Just updated {away_team.abbrev} ranking to {away_team.ranking}')
logging.info(f"Just updated {away_team.abbrev} ranking to {away_team.ranking}")
home_team = Team.get_by_id(result.home_team_id)
home_team.ranking += home_delta
home_team.save()
logging.info(f'Just updated {home_team.abbrev} ranking to {home_team.ranking}')
logging.info(f"Just updated {home_team.abbrev} ranking to {home_team.ranking}")
if saved == 1:
return_val = model_to_dict(this_result)
@ -313,27 +410,38 @@ async def post_result(result: ResultModel, token: str = Depends(oauth2_scheme)):
else:
raise HTTPException(
status_code=418,
detail='Well slap my ass and call me a teapot; I could not save that roster'
detail="Well slap my ass and call me a teapot; I could not save that roster",
)
@router.patch('/{result_id}')
@router.patch("/{result_id}")
async def patch_result(
result_id, away_team_id: Optional[int] = None, home_team_id: Optional[int] = None,
away_score: Optional[int] = None, home_score: Optional[int] = None, away_team_value: Optional[int] = None,
home_team_value: Optional[int] = None, scorecard: Optional[str] = None, week: Optional[int] = None,
season: Optional[int] = None, short_game: Optional[bool] = None, game_type: Optional[str] = None,
token: str = Depends(oauth2_scheme)):
result_id,
away_team_id: Optional[int] = None,
home_team_id: Optional[int] = None,
away_score: Optional[int] = None,
home_score: Optional[int] = None,
away_team_value: Optional[int] = None,
home_team_value: Optional[int] = None,
scorecard: Optional[str] = None,
week: Optional[int] = None,
season: Optional[int] = None,
short_game: Optional[bool] = None,
game_type: Optional[str] = None,
token: str = Depends(oauth2_scheme),
):
if not valid_token(token):
logging.warning('Bad Token: [REDACTED]')
logging.warning("Bad Token: [REDACTED]")
raise HTTPException(
status_code=401,
detail='You are not authorized to patch results. This event has been logged.'
detail="You are not authorized to patch results. This event has been logged.",
)
try:
this_result = Result.get_by_id(result_id)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No result found with id {result_id}')
raise HTTPException(
status_code=404, detail=f"No result found with id {result_id}"
)
if away_team_id is not None:
this_result.away_team_id = away_team_id
@ -377,27 +485,32 @@ async def patch_result(
else:
raise HTTPException(
status_code=418,
detail='Well slap my ass and call me a teapot; I could not save that event'
detail="Well slap my ass and call me a teapot; I could not save that event",
)
@router.delete('/{result_id}')
@router.delete("/{result_id}")
async def delete_result(result_id, token: str = Depends(oauth2_scheme)):
if not valid_token(token):
logging.warning('Bad Token: [REDACTED]')
logging.warning("Bad Token: [REDACTED]")
raise HTTPException(
status_code=401,
detail='You are not authorized to post results. This event has been logged.'
detail="You are not authorized to post results. This event has been logged.",
)
try:
this_result = Result.get_by_id(result_id)
except DoesNotExist:
raise HTTPException(status_code=404, detail=f'No result found with id {result_id}')
raise HTTPException(
status_code=404, detail=f"No result found with id {result_id}"
)
count = this_result.delete_instance()
if count == 1:
raise HTTPException(status_code=200, detail=f'Result {result_id} has been deleted')
raise HTTPException(
status_code=200, detail=f"Result {result_id} has been deleted"
)
else:
raise HTTPException(status_code=500, detail=f'Result {result_id} was not deleted')
raise HTTPException(
status_code=500, detail=f"Result {result_id} was not deleted"
)