cal/major-domo-database
1
0
Fork 0
Code Issues 25 Pull Requests 25 Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix: assign order_by() return value in GET /api/v3/games (#24)' (#39) from ai/major-domo-database-24 into main
All checks were successful
Build Docker Image / build (push) Successful in 2m9s
Details

Browse Source 
Reviewed-on: #39
This commit is contained in:
cal 2026-03-03 01:29:27 +00:00
commit a0d27031b4
1 changed files with 119 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

178
app/routers_v3/stratgame.py
View File

@ -5,21 +5,26 @@ import logging
import pydantic import pydantic
from ..db_engine import db, StratGame, Team, StratPlay, model_to_dict, chunked, fn from ..db_engine import db, StratGame, Team, StratPlay, model_to_dict, chunked, fn
from ..dependencies import oauth2_scheme, send_webhook_message, update_season_pitching_stats, valid_token, PRIVATE_IN_SCHEMA, handle_db_errors, update_season_batting_stats from ..dependencies import (
oauth2_scheme,
logger = logging.getLogger('discord_app') send_webhook_message,
update_season_pitching_stats,
router = APIRouter( valid_token,
prefix='/api/v3/games', PRIVATE_IN_SCHEMA,
tags=['games'] handle_db_errors,
update_season_batting_stats,
) )
logger = logging.getLogger("discord_app")
router = APIRouter(prefix="/api/v3/games", tags=["games"])
class GameModel(pydantic.BaseModel): class GameModel(pydantic.BaseModel):
season: int season: int
week: int week: int
game_num: Optional[int] = None game_num: Optional[int] = None
season_type: Optional[str] = 'regular' season_type: Optional[str] = "regular"
away_team_id: int away_team_id: int
home_team_id: int home_team_id: int
away_score: Optional[int] = None away_score: Optional[int] = None
@ -33,16 +38,28 @@ class GameList(pydantic.BaseModel):
games: List[GameModel] games: List[GameModel]
@router.get('') @router.get("")
@handle_db_errors @handle_db_errors
async def get_games( async def get_games(
season: list = Query(default=None), week: list = Query(default=None), game_num: list = Query(default=None), season: list = Query(default=None),
season_type: Literal['regular', 'post', 'all'] = 'all', away_team_id: list = Query(default=None), week: list = Query(default=None),
home_team_id: list = Query(default=None), week_start: Optional[int] = None, week_end: Optional[int] = None, game_num: list = Query(default=None),
team1_id: list = Query(default=None), team2_id: list = Query(default=None), played: Optional[bool] = None, season_type: Literal["regular", "post", "all"] = "all",
away_manager_id: list = Query(default=None), home_manager_id: list = Query(default=None), away_team_id: list = Query(default=None),
manager1_id: list = Query(default=None), manager2_id: list = Query(default=None), home_team_id: list = Query(default=None),
division_id: Optional[int] = None, short_output: Optional[bool] = False, sort: Optional[str] = None) -> Any: week_start: Optional[int] = None,
week_end: Optional[int] = None,
team1_id: list = Query(default=None),
team2_id: list = Query(default=None),
played: Optional[bool] = None,
away_manager_id: list = Query(default=None),
home_manager_id: list = Query(default=None),
manager1_id: list = Query(default=None),
manager2_id: list = Query(default=None),
division_id: Optional[int] = None,
short_output: Optional[bool] = False,
sort: Optional[str] = None,
) -> Any:
all_games = StratGame.select() all_games = StratGame.select()
if season is not None: if season is not None:
@ -51,7 +68,7 @@ async def get_games(
all_games = all_games.where(StratGame.week << week) all_games = all_games.where(StratGame.week << week)
if game_num is not None: if game_num is not None:
all_games = all_games.where(StratGame.game_num << game_num) all_games = all_games.where(StratGame.game_num << game_num)
if season_type != 'all': if season_type != "all":
all_games = all_games.where(StratGame.season_type == season_type) all_games = all_games.where(StratGame.season_type == season_type)
if away_team_id is not None: if away_team_id is not None:
all_games = all_games.where(StratGame.away_team_id << away_team_id) all_games = all_games.where(StratGame.away_team_id << away_team_id)
@ -80,57 +97,69 @@ async def get_games(
all_games = all_games.where(StratGame.home_manager_id << home_manager_id) all_games = all_games.where(StratGame.home_manager_id << home_manager_id)
if manager1_id is not None: if manager1_id is not None:
all_games = all_games.where( all_games = all_games.where(
(StratGame.away_manager_id << manager1_id) | (StratGame.home_manager_id << manager1_id) (StratGame.away_manager_id << manager1_id)
| (StratGame.home_manager_id << manager1_id)
) )
if manager2_id is not None: if manager2_id is not None:
all_games = all_games.where( all_games = all_games.where(
(StratGame.away_manager_id << manager2_id) | (StratGame.home_manager_id << manager2_id) (StratGame.away_manager_id << manager2_id)
| (StratGame.home_manager_id << manager2_id)
) )
if played is not None: if played is not None:
all_games = all_games.where(StratGame.game_num.is_null(not played)) all_games = all_games.where(StratGame.game_num.is_null(not played))
if game_num is not None: if game_num is not None:
all_games = all_games.where(StratGame.game_num << game_num) all_games = all_games.where(StratGame.game_num << game_num)
if sort == 'recent-first': if sort == "recent-first":
all_games.order_by(-StratGame.season, -StratGame.week, -StratGame.game_num) all_games = all_games.order_by(
-StratGame.season, -StratGame.week, -StratGame.game_num
)
else: else:
all_games.order_by(StratGame.season, StratGame.week, StratGame.game_num) all_games = all_games.order_by(
StratGame.season, StratGame.week, StratGame.game_num
)
return_games = { return_games = {
'count': all_games.count(), "count": all_games.count(),
'games': [model_to_dict(x, recurse=not short_output) for x in all_games] "games": [model_to_dict(x, recurse=not short_output) for x in all_games],
} }
db.close() db.close()
return return_games return return_games
@router.get('/{game_id}') @router.get("/{game_id}")
@handle_db_errors @handle_db_errors
async def get_one_game(game_id: int) -> Any: async def get_one_game(game_id: int) -> Any:
this_game = StratGame.get_or_none(StratGame.id == game_id) this_game = StratGame.get_or_none(StratGame.id == game_id)
if not this_game: if not this_game:
db.close() db.close()
raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found') raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
g_result = model_to_dict(this_game) g_result = model_to_dict(this_game)
db.close() db.close()
return g_result return g_result
@router.patch('/{game_id}', include_in_schema=PRIVATE_IN_SCHEMA) @router.patch("/{game_id}", include_in_schema=PRIVATE_IN_SCHEMA)
@handle_db_errors @handle_db_errors
async def patch_game( async def patch_game(
game_id: int, game_num: Optional[int] = None, away_score: Optional[int] = None, game_id: int,
home_score: Optional[int] = None, away_manager_id: Optional[int] = None, home_manager_id: Optional[int] = None, game_num: Optional[int] = None,
token: str = Depends(oauth2_scheme), scorecard_url: Optional[str] = None) -> Any: away_score: Optional[int] = None,
home_score: Optional[int] = None,
away_manager_id: Optional[int] = None,
home_manager_id: Optional[int] = None,
token: str = Depends(oauth2_scheme),
scorecard_url: Optional[str] = None,
) -> Any:
if not valid_token(token): if not valid_token(token):
logger.warning(f'patch_game - Bad Token: {token}') logger.warning(f"patch_game - Bad Token: {token}")
raise HTTPException(status_code=401, detail='Unauthorized') raise HTTPException(status_code=401, detail="Unauthorized")
this_game = StratGame.get_or_none(StratGame.id == game_id) this_game = StratGame.get_or_none(StratGame.id == game_id)
if not this_game: if not this_game:
db.close() db.close()
raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found') raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
if game_num is not None: if game_num is not None:
this_game.game_num = game_num this_game.game_num = game_num
@ -149,55 +178,75 @@ async def patch_game(
# Update batting stats for all batters in this game # Update batting stats for all batters in this game
try: try:
# Get all unique batter IDs from stratplays in this game # Get all unique batter IDs from stratplays in this game
batter_ids = [row.batter_id for row in StratPlay.select(StratPlay.batter_id.distinct()) batter_ids = [
.where(StratPlay.game_id == game_id)] row.batter_id
for row in StratPlay.select(StratPlay.batter_id.distinct()).where(
StratPlay.game_id == game_id
)
]
if batter_ids: if batter_ids:
update_season_batting_stats(batter_ids, this_game.season, db) update_season_batting_stats(batter_ids, this_game.season, db)
logger.info(f'Updated batting stats for {len(batter_ids)} players from game {game_id}') logger.info(
f"Updated batting stats for {len(batter_ids)} players from game {game_id}"
)
else: else:
logger.error(f'No batters found for game_id {game_id}') logger.error(f"No batters found for game_id {game_id}")
except Exception as e: except Exception as e:
logger.error(f'Failed to update batting stats for game {game_id}: {e}') logger.error(f"Failed to update batting stats for game {game_id}: {e}")
send_webhook_message(f'Failed to update season batting stats after game {this_game.game_num} between {this_game.away_team.abbrev} and {this_game.home_team.abbrev}!\nError: {e}') send_webhook_message(
f"Failed to update season batting stats after game {this_game.game_num} between {this_game.away_team.abbrev} and {this_game.home_team.abbrev}!\nError: {e}"
)
# Don't fail the patch operation if stats update fails # Don't fail the patch operation if stats update fails
# Update pitching stats for all pitchers in this game # Update pitching stats for all pitchers in this game
try: try:
# Get all unique pitcher IDs from stratplays in this game # Get all unique pitcher IDs from stratplays in this game
pitcher_ids = [row.pitcher_id for row in StratPlay.select(StratPlay.pitcher_id.distinct()) pitcher_ids = [
.where(StratPlay.game_id == game_id)] row.pitcher_id
for row in StratPlay.select(StratPlay.pitcher_id.distinct()).where(
StratPlay.game_id == game_id
)
]
if pitcher_ids: if pitcher_ids:
update_season_pitching_stats(pitcher_ids, this_game.season, db) update_season_pitching_stats(pitcher_ids, this_game.season, db)
logger.info(f'Updated pitching stats for {len(pitcher_ids)} players from game {game_id}') logger.info(
f"Updated pitching stats for {len(pitcher_ids)} players from game {game_id}"
)
else: else:
logger.error(f'No pitchers found for game_id {game_id}') logger.error(f"No pitchers found for game_id {game_id}")
except Exception as e: except Exception as e:
logger.error(f'Failed to update pitching stats for game {game_id}: {e}') logger.error(f"Failed to update pitching stats for game {game_id}: {e}")
send_webhook_message(f'Failed to update season pitching stats after game {this_game.game_num} between {this_game.away_team.abbrev} and {this_game.home_team.abbrev}!\nError: {e}') send_webhook_message(
f"Failed to update season pitching stats after game {this_game.game_num} between {this_game.away_team.abbrev} and {this_game.home_team.abbrev}!\nError: {e}"
)
g_result = model_to_dict(this_game) g_result = model_to_dict(this_game)
return g_result return g_result
else: else:
raise HTTPException(status_code=500, detail=f'Unable to patch game {game_id}') raise HTTPException(status_code=500, detail=f"Unable to patch game {game_id}")
@router.post('', include_in_schema=PRIVATE_IN_SCHEMA) @router.post("", include_in_schema=PRIVATE_IN_SCHEMA)
@handle_db_errors @handle_db_errors
async def post_games(game_list: GameList, token: str = Depends(oauth2_scheme)) -> Any: async def post_games(game_list: GameList, token: str = Depends(oauth2_scheme)) -> Any:
if not valid_token(token): if not valid_token(token):
logger.warning(f'post_games - Bad Token: {token}') logger.warning(f"post_games - Bad Token: {token}")
raise HTTPException(status_code=401, detail='Unauthorized') raise HTTPException(status_code=401, detail="Unauthorized")
new_games = [] new_games = []
for x in game_list.games: for x in game_list.games:
if Team.get_or_none(Team.id == x.away_team_id) is None: if Team.get_or_none(Team.id == x.away_team_id) is None:
raise HTTPException(status_code=404, detail=f'Team ID {x.away_team_id} not found') raise HTTPException(
status_code=404, detail=f"Team ID {x.away_team_id} not found"
)
if Team.get_or_none(Team.id == x.home_team_id) is None: if Team.get_or_none(Team.id == x.home_team_id) is None:
raise HTTPException(status_code=404, detail=f'Team ID {x.home_team_id} not found') raise HTTPException(
status_code=404, detail=f"Team ID {x.home_team_id} not found"
)
new_games.append(x.dict()) new_games.append(x.dict())
@ -206,20 +255,20 @@ async def post_games(game_list: GameList, token: str = Depends(oauth2_scheme)) -
StratGame.insert_many(batch).on_conflict_ignore().execute() StratGame.insert_many(batch).on_conflict_ignore().execute()
db.close() db.close()
return f'Inserted {len(new_games)} games' return f"Inserted {len(new_games)} games"
@router.post('/wipe/{game_id}', include_in_schema=PRIVATE_IN_SCHEMA) @router.post("/wipe/{game_id}", include_in_schema=PRIVATE_IN_SCHEMA)
@handle_db_errors @handle_db_errors
async def wipe_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any: async def wipe_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any:
if not valid_token(token): if not valid_token(token):
logger.warning(f'wipe_game - Bad Token: {token}') logger.warning(f"wipe_game - Bad Token: {token}")
raise HTTPException(status_code=401, detail='Unauthorized') raise HTTPException(status_code=401, detail="Unauthorized")
this_game = StratGame.get_or_none(StratGame.id == game_id) this_game = StratGame.get_or_none(StratGame.id == game_id)
if not this_game: if not this_game:
db.close() db.close()
raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found') raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
this_game.away_score = None this_game.away_score = None
this_game.home_score = None this_game.home_score = None
@ -233,26 +282,27 @@ async def wipe_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any:
return g_result return g_result
else: else:
db.close() db.close()
raise HTTPException(status_code=500, detail=f'Unable to wipe game {game_id}') raise HTTPException(status_code=500, detail=f"Unable to wipe game {game_id}")
@router.delete('/{game_id}', include_in_schema=PRIVATE_IN_SCHEMA) @router.delete("/{game_id}", include_in_schema=PRIVATE_IN_SCHEMA)
@handle_db_errors @handle_db_errors
async def delete_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any: async def delete_game(game_id: int, token: str = Depends(oauth2_scheme)) -> Any:
if not valid_token(token): if not valid_token(token):
logger.warning(f'delete_game - Bad Token: {token}') logger.warning(f"delete_game - Bad Token: {token}")
raise HTTPException(status_code=401, detail='Unauthorized') raise HTTPException(status_code=401, detail="Unauthorized")
this_game = StratGame.get_or_none(StratGame.id == game_id) this_game = StratGame.get_or_none(StratGame.id == game_id)
if not this_game: if not this_game:
db.close() db.close()
raise HTTPException(status_code=404, detail=f'StratGame ID {game_id} not found') raise HTTPException(status_code=404, detail=f"StratGame ID {game_id} not found")
count = this_game.delete_instance() count = this_game.delete_instance()
db.close() db.close()
if count == 1: if count == 1:
return f'StratGame {game_id} has been deleted' return f"StratGame {game_id} has been deleted"
else: else:
raise HTTPException(status_code=500, detail=f'StratGame {game_id} could not be deleted') raise HTTPException(
status_code=500, detail=f"StratGame {game_id} could not be deleted"
)