16f3f8d8de
All checks were successful
Build Docker Image / build (pull_request) Successful in 2m32s
Add MAX_LIMIT=500 cap across all list endpoints, empty string stripping middleware, and limit/offset to /transactions. Resolves #98. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
215 lines
8.2 KiB
Python
215 lines
8.2 KiB
Python
import logging
|
|
from typing import Optional, Literal
|
|
|
|
from fastapi import APIRouter, Query
|
|
|
|
from ...db_engine import (
|
|
db,
|
|
StratPlay,
|
|
StratGame,
|
|
Team,
|
|
Player,
|
|
model_to_dict,
|
|
fn,
|
|
)
|
|
from ...dependencies import (
|
|
handle_db_errors,
|
|
add_cache_headers,
|
|
cache_result,
|
|
MAX_LIMIT,
|
|
DEFAULT_LIMIT,
|
|
)
|
|
|
|
logger = logging.getLogger("discord_app")
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/")
|
|
@handle_db_errors
|
|
@add_cache_headers(max_age=10 * 60)
|
|
@cache_result(ttl=5 * 60, key_prefix="plays")
|
|
async def get_plays(
|
|
game_id: list = Query(default=None),
|
|
batter_id: list = Query(default=None),
|
|
season: list = Query(default=None),
|
|
week: list = Query(default=None),
|
|
has_defender: Optional[bool] = None,
|
|
has_catcher: Optional[bool] = None,
|
|
has_defender_or_catcher: Optional[bool] = None,
|
|
is_scoring_play: Optional[bool] = None,
|
|
pitcher_id: list = Query(default=None),
|
|
obc: list = Query(default=None),
|
|
inning: list = Query(default=None),
|
|
batting_order: list = Query(default=None),
|
|
starting_outs: list = Query(default=None),
|
|
batter_pos: list = Query(default=None),
|
|
catcher_id: list = Query(default=None),
|
|
defender_id: list = Query(default=None),
|
|
runner_id: list = Query(default=None),
|
|
offense_team_id: list = Query(default=None),
|
|
defense_team_id: list = Query(default=None),
|
|
hit: Optional[int] = None,
|
|
double: Optional[int] = None,
|
|
triple: Optional[int] = None,
|
|
homerun: Optional[int] = None,
|
|
play_num: list = Query(default=None),
|
|
error: list = Query(default=None),
|
|
sb: Optional[int] = None,
|
|
cs: Optional[int] = None,
|
|
manager_id: list = Query(default=None),
|
|
run: Optional[int] = None,
|
|
e_run: Optional[int] = None,
|
|
rbi: list = Query(default=None),
|
|
outs: list = Query(default=None),
|
|
wild_pitch: Optional[int] = None,
|
|
is_final_out: Optional[bool] = None,
|
|
is_go_ahead: Optional[bool] = None,
|
|
is_tied: Optional[bool] = None,
|
|
is_new_inning: Optional[bool] = None,
|
|
min_wpa: Optional[float] = None,
|
|
max_wpa: Optional[float] = None,
|
|
pitcher_team_id: list = Query(default=None),
|
|
short_output: Optional[bool] = False,
|
|
sort: Optional[str] = None,
|
|
limit: int = Query(default=DEFAULT_LIMIT, ge=1, le=MAX_LIMIT),
|
|
page_num: Optional[int] = 1,
|
|
s_type: Literal["regular", "post", "total", None] = None,
|
|
):
|
|
all_plays = StratPlay.select()
|
|
|
|
if season is not None:
|
|
s_games = StratGame.select().where(StratGame.season << season)
|
|
all_plays = all_plays.where(StratPlay.game << s_games)
|
|
if week is not None:
|
|
w_games = StratGame.select().where(StratGame.week << week)
|
|
all_plays = all_plays.where(StratPlay.game << w_games)
|
|
if has_defender is not None:
|
|
all_plays = all_plays.where(StratPlay.defender.is_null(False))
|
|
if has_catcher is not None:
|
|
all_plays = all_plays.where(StratPlay.catcher.is_null(False))
|
|
if has_defender_or_catcher is not None:
|
|
all_plays = all_plays.where(
|
|
(StratPlay.catcher.is_null(False)) | (StratPlay.defender.is_null(False))
|
|
)
|
|
if game_id is not None:
|
|
all_plays = all_plays.where(StratPlay.game_id << game_id)
|
|
if batter_id is not None:
|
|
all_plays = all_plays.where(StratPlay.batter_id << batter_id)
|
|
if pitcher_id is not None:
|
|
all_plays = all_plays.where(StratPlay.pitcher_id << pitcher_id)
|
|
if obc is not None:
|
|
all_plays = all_plays.where(StratPlay.on_base_code << obc)
|
|
if inning is not None:
|
|
all_plays = all_plays.where(StratPlay.inning_num << inning)
|
|
if batting_order is not None:
|
|
all_plays = all_plays.where(StratPlay.batting_order << batting_order)
|
|
if starting_outs is not None:
|
|
all_plays = all_plays.where(StratPlay.starting_outs << starting_outs)
|
|
if batter_pos is not None:
|
|
all_plays = all_plays.where(StratPlay.batter_pos << batter_pos)
|
|
if catcher_id is not None:
|
|
all_plays = all_plays.where(StratPlay.catcher_id << catcher_id)
|
|
if defender_id is not None:
|
|
all_plays = all_plays.where(StratPlay.defender_id << defender_id)
|
|
if runner_id is not None:
|
|
all_plays = all_plays.where(StratPlay.runner_id << runner_id)
|
|
if pitcher_team_id is not None:
|
|
all_teams = Team.select().where(Team.id << pitcher_team_id)
|
|
all_plays = all_plays.where((StratPlay.pitcher_team << all_teams))
|
|
if offense_team_id is not None:
|
|
all_teams = Team.select().where(Team.id << offense_team_id)
|
|
all_plays = all_plays.where(
|
|
(StratPlay.batter_team << all_teams) | (StratPlay.runner_team << all_teams)
|
|
)
|
|
if defense_team_id is not None:
|
|
all_teams = Team.select().where(Team.id << defense_team_id)
|
|
all_plays = all_plays.where(
|
|
(StratPlay.catcher_team << all_teams)
|
|
| (StratPlay.defender_team << all_teams)
|
|
)
|
|
if hit is not None:
|
|
all_plays = all_plays.where(StratPlay.hit == hit)
|
|
if double is not None:
|
|
all_plays = all_plays.where(StratPlay.double == double)
|
|
if triple is not None:
|
|
all_plays = all_plays.where(StratPlay.triple == triple)
|
|
if homerun is not None:
|
|
all_plays = all_plays.where(StratPlay.homerun == homerun)
|
|
if sb is not None:
|
|
all_plays = all_plays.where(StratPlay.sb == sb)
|
|
if cs is not None:
|
|
all_plays = all_plays.where(StratPlay.cs == cs)
|
|
if wild_pitch is not None:
|
|
all_plays = all_plays.where(StratPlay.wild_pitch == wild_pitch)
|
|
if run is not None:
|
|
all_plays = all_plays.where(StratPlay.run == run)
|
|
if e_run is not None:
|
|
all_plays = all_plays.where(StratPlay.e_run == e_run)
|
|
if rbi is not None:
|
|
all_plays = all_plays.where(StratPlay.rbi << rbi)
|
|
if outs is not None:
|
|
all_plays = all_plays.where(StratPlay.outs << outs)
|
|
if error is not None:
|
|
all_plays = all_plays.where(StratPlay.error << error)
|
|
if manager_id is not None:
|
|
all_games = StratGame.select().where(
|
|
(StratGame.away_manager_id << manager_id)
|
|
| (StratGame.home_manager_id << manager_id)
|
|
)
|
|
all_plays = all_plays.where(StratPlay.game << all_games)
|
|
if is_final_out is not None:
|
|
all_plays = all_plays.where(StratPlay.starting_outs + StratPlay.outs == 3)
|
|
if is_go_ahead is not None:
|
|
all_plays = all_plays.where(StratPlay.is_go_ahead == is_go_ahead)
|
|
if is_tied is not None:
|
|
all_plays = all_plays.where(StratPlay.is_tied == is_tied)
|
|
if is_new_inning is not None:
|
|
all_plays = all_plays.where(StratPlay.is_new_inning == is_new_inning)
|
|
if is_scoring_play is not None:
|
|
all_plays = all_plays.where(
|
|
(StratPlay.on_first_final == 4)
|
|
| (StratPlay.on_second_final == 4)
|
|
| (StratPlay.on_third_final == 4)
|
|
| (StratPlay.batter_final == 4)
|
|
)
|
|
if min_wpa is not None:
|
|
all_plays = all_plays.where(StratPlay.wpa >= min_wpa)
|
|
if max_wpa is not None:
|
|
all_plays = all_plays.where(StratPlay.wpa <= max_wpa)
|
|
if play_num is not None:
|
|
all_plays = all_plays.where(StratPlay.play_num << play_num)
|
|
if s_type is not None:
|
|
season_games = StratGame.select()
|
|
if s_type == "regular":
|
|
season_games = season_games.where(StratGame.week <= 18)
|
|
elif s_type == "post":
|
|
season_games = season_games.where(StratGame.week > 18)
|
|
all_plays = all_plays.where(StratPlay.game << season_games)
|
|
|
|
bat_plays = all_plays.paginate(page_num, limit)
|
|
|
|
if sort == "wpa-desc":
|
|
all_plays = all_plays.order_by(-fn.ABS(StratPlay.wpa))
|
|
elif sort == "wpa-asc":
|
|
all_plays = all_plays.order_by(fn.ABS(StratPlay.wpa))
|
|
elif sort == "re24-desc":
|
|
all_plays = all_plays.order_by(-fn.ABS(StratPlay.re24_primary))
|
|
elif sort == "re24-asc":
|
|
all_plays = all_plays.order_by(fn.ABS(StratPlay.re24_primary))
|
|
elif sort == "newest":
|
|
all_plays = all_plays.order_by(
|
|
StratPlay.game_id.desc(), StratPlay.play_num.desc()
|
|
)
|
|
elif sort == "oldest":
|
|
all_plays = all_plays.order_by(StratPlay.game_id, StratPlay.play_num)
|
|
|
|
all_plays = all_plays.limit(limit)
|
|
|
|
return_plays = {
|
|
"count": all_plays.count(),
|
|
"plays": [model_to_dict(x, recurse=not short_output) for x in all_plays],
|
|
}
|
|
db.close()
|
|
return return_plays
|