701098881a
- Add validation in create_game() and quick_create_game() to ensure both teams are successfully fetched from SBA API before creating game - Raise HTTP 400 with clear error message if team data cannot be fetched - Add warning logs in get_teams_by_ids() when teams are missing from result - Prevents games from being created with null team display info (names, abbreviations, colors, thumbnails) Root cause: get_teams_by_ids() silently returned empty dict on API failures, and game creation endpoints didn't validate the result. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
395 lines
13 KiB
Python
395 lines
13 KiB
Python
"""
|
|
SBA API client for fetching player and team data.
|
|
|
|
Integrates with SBA REST API to retrieve player and team information
|
|
for use in game lineup display and game creation.
|
|
|
|
Author: Claude
|
|
Date: 2025-01-10
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pendulum
|
|
|
|
from app.config import get_settings
|
|
from app.models.player_models import SbaPlayer
|
|
|
|
logger = logging.getLogger(f"{__name__}.SbaApiClient")
|
|
settings = get_settings()
|
|
|
|
# Module-level team cache (team_id -> team_data)
|
|
_team_cache: dict[int, dict[str, Any]] = {}
|
|
_team_cache_expiry: pendulum.DateTime | None = None
|
|
_CACHE_TTL_HOURS = 1
|
|
|
|
|
|
class SbaApiClient:
|
|
"""Client for SBA API player and team data lookups."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str = "https://api.sba.manticorum.com",
|
|
api_key: str | None = None,
|
|
):
|
|
"""
|
|
Initialize SBA API client.
|
|
|
|
Args:
|
|
base_url: Base URL for SBA API (default: production)
|
|
api_key: Bearer token for API authentication
|
|
"""
|
|
self.base_url = base_url
|
|
self.api_key = api_key or settings.sba_api_key
|
|
self.timeout = httpx.Timeout(10.0, connect=5.0)
|
|
|
|
def _get_headers(self) -> dict[str, str]:
|
|
"""Get headers with auth token."""
|
|
headers = {}
|
|
if self.api_key:
|
|
headers["Authorization"] = f"Bearer {self.api_key}"
|
|
return headers
|
|
|
|
async def get_teams(self, season: int, active_only: bool = True) -> list[dict[str, Any]]:
|
|
"""
|
|
Fetch teams from SBA API for a specific season.
|
|
|
|
Args:
|
|
season: Season number (e.g., 3 for Season 3)
|
|
active_only: If True, filter out IL teams and teams without divisions
|
|
|
|
Returns:
|
|
List of team dictionaries with id, name, abbreviation, etc.
|
|
|
|
Example:
|
|
teams = await client.get_teams(season=3)
|
|
for team in teams:
|
|
print(f"{team['abbrev']}: {team['lname']}")
|
|
"""
|
|
url = f"{self.base_url}/teams"
|
|
params = {"season": season}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
url, headers=self._get_headers(), params=params
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
teams = data.get("teams", [])
|
|
|
|
if active_only:
|
|
# Filter out injured list teams (IL suffix) and teams without divisions
|
|
teams = [
|
|
t for t in teams
|
|
if not t["abbrev"].endswith("IL") and t.get("division")
|
|
]
|
|
|
|
logger.info(f"Loaded {len(teams)} teams for season {season}")
|
|
return teams
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Failed to fetch teams: {e}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error fetching teams: {e}")
|
|
raise
|
|
|
|
async def get_teams_by_owner(self, discord_id: str, season: int) -> list[dict[str, Any]]:
|
|
"""
|
|
Fetch teams owned by a specific Discord user.
|
|
|
|
Args:
|
|
discord_id: Discord user ID (gmid or gmid2)
|
|
season: Season number
|
|
|
|
Returns:
|
|
List of team dictionaries owned by this user
|
|
|
|
Example:
|
|
teams = await client.get_teams_by_owner("485217045408120833", season=13)
|
|
"""
|
|
url = f"{self.base_url}/teams"
|
|
params = {"season": season, "owner_id": discord_id}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
url, headers=self._get_headers(), params=params
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
teams = data.get("teams", [])
|
|
|
|
logger.info(f"Found {len(teams)} teams for owner {discord_id[:8]}...")
|
|
return teams
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Failed to fetch teams for owner {discord_id[:8]}...: {e}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error fetching teams by owner: {e}")
|
|
raise
|
|
|
|
async def get_team_by_id(self, team_id: int, season: int = 3) -> dict[str, Any] | None:
|
|
"""
|
|
Get a single team by ID, using cache when available.
|
|
|
|
Args:
|
|
team_id: Team ID to look up
|
|
season: Season number (default: 3)
|
|
|
|
Returns:
|
|
Team dictionary with id, lname, abbrev, etc. or None if not found
|
|
|
|
Example:
|
|
team = await client.get_team_by_id(35)
|
|
print(team['lname']) # "Cardinals"
|
|
"""
|
|
global _team_cache, _team_cache_expiry
|
|
|
|
now = pendulum.now("UTC")
|
|
|
|
# Check if cache is valid
|
|
if _team_cache_expiry is None or now > _team_cache_expiry:
|
|
# Refresh cache
|
|
try:
|
|
teams = await self.get_teams(season=season, active_only=False)
|
|
_team_cache = {t["id"]: t for t in teams}
|
|
_team_cache_expiry = now.add(hours=_CACHE_TTL_HOURS)
|
|
logger.info(f"Refreshed team cache with {len(_team_cache)} teams")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to refresh team cache: {e}")
|
|
# Continue with stale cache if available
|
|
|
|
return _team_cache.get(team_id)
|
|
|
|
async def get_teams_by_ids(self, team_ids: list[int], season: int = 3) -> dict[int, dict[str, Any]]:
|
|
"""
|
|
Get multiple teams by ID, using cache.
|
|
|
|
Args:
|
|
team_ids: List of team IDs to look up
|
|
season: Season number (default: 3)
|
|
|
|
Returns:
|
|
Dictionary mapping team_id to team data
|
|
"""
|
|
result = {}
|
|
for team_id in team_ids:
|
|
team = await self.get_team_by_id(team_id, season)
|
|
if team:
|
|
result[team_id] = team
|
|
else:
|
|
logger.warning(f"Team {team_id} not found in cache for season {season}")
|
|
|
|
# Log if we couldn't find all requested teams
|
|
missing = [tid for tid in team_ids if tid not in result]
|
|
if missing:
|
|
logger.warning(
|
|
f"get_teams_by_ids: {len(missing)}/{len(team_ids)} teams not found: {missing}"
|
|
)
|
|
|
|
return result
|
|
|
|
async def get_player(self, player_id: int) -> SbaPlayer:
|
|
"""
|
|
Fetch player data from SBA API.
|
|
|
|
Args:
|
|
player_id: SBA player ID
|
|
|
|
Returns:
|
|
SbaPlayer instance with all player data
|
|
|
|
Raises:
|
|
httpx.HTTPError: If API request fails
|
|
|
|
Example:
|
|
player = await client.get_player(12288)
|
|
print(f"Name: {player.name}") # "Ronald Acuna Jr"
|
|
print(f"Positions: {player.get_positions()}") # ['RF']
|
|
"""
|
|
url = f"{self.base_url}/players/{player_id}"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(url, headers=self._get_headers())
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
player = SbaPlayer.from_api_response(data)
|
|
|
|
logger.info(f"Loaded player {player_id}: {player.name}")
|
|
return player
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Failed to fetch player {player_id}: {e}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error fetching player {player_id}: {e}")
|
|
raise
|
|
|
|
async def get_roster(self, team_id: int, season: int) -> list[dict[str, Any]]:
|
|
"""
|
|
Fetch roster for a specific team from SBA API.
|
|
|
|
Args:
|
|
team_id: Team ID
|
|
season: Season number (e.g., 3 for Season 3)
|
|
|
|
Returns:
|
|
List of player dictionaries from the SBA API
|
|
|
|
Example:
|
|
roster = await client.get_roster(team_id=35, season=3)
|
|
for player in roster:
|
|
print(f"{player['name']}: {player['position']}")
|
|
"""
|
|
url = f"{self.base_url}/players"
|
|
params = {"season": season, "team_id": team_id}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
url, headers=self._get_headers(), params=params
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
players = data.get("players", [])
|
|
count = data.get("count", len(players))
|
|
|
|
logger.info(f"Loaded {count} players for team {team_id}, season {season}")
|
|
return players
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Failed to fetch roster for team {team_id}: {e}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error fetching roster: {e}")
|
|
raise
|
|
|
|
async def get_players_batch(self, player_ids: list[int]) -> dict[int, SbaPlayer]:
|
|
"""
|
|
Fetch multiple players in parallel.
|
|
|
|
Args:
|
|
player_ids: List of SBA player IDs to fetch
|
|
|
|
Returns:
|
|
Dictionary mapping player_id to SbaPlayer instance
|
|
Players that fail to load will be omitted from the result
|
|
|
|
Example:
|
|
players = await client.get_players_batch([12288, 12289, 12290])
|
|
for player_id, player in players.items():
|
|
print(f"{player.name}: {player.get_positions()}")
|
|
"""
|
|
if not player_ids:
|
|
return {}
|
|
|
|
results: dict[int, SbaPlayer] = {}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
for player_id in player_ids:
|
|
try:
|
|
url = f"{self.base_url}/players/{player_id}"
|
|
response = await client.get(url)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
player = SbaPlayer.from_api_response(data)
|
|
results[player_id] = player
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.warning(f"Failed to fetch player {player_id}: {e}")
|
|
# Continue with other players
|
|
except Exception as e:
|
|
logger.warning(f"Unexpected error fetching player {player_id}: {e}")
|
|
# Continue with other players
|
|
|
|
logger.info(f"Loaded {len(results)}/{len(player_ids)} players")
|
|
return results
|
|
|
|
async def get_current(self) -> dict[str, Any]:
|
|
"""
|
|
Get current season and week from SBA API.
|
|
|
|
Returns:
|
|
Dictionary with 'season' and 'week' keys
|
|
|
|
Example:
|
|
current = await client.get_current()
|
|
print(f"Season {current['season']}, Week {current['week']}")
|
|
"""
|
|
url = f"{self.base_url}/current"
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(url, headers=self._get_headers())
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
logger.info(f"Current: Season {data.get('season')}, Week {data.get('week')}")
|
|
return data
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Failed to fetch current season/week: {e}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error fetching current: {e}")
|
|
raise
|
|
|
|
async def get_schedule_games(
|
|
self, season: int, week_start: int, week_end: int
|
|
) -> list[dict[str, Any]]:
|
|
"""
|
|
Get scheduled games for a week range from SBA API.
|
|
|
|
Args:
|
|
season: Season number (e.g., 13)
|
|
week_start: Starting week number
|
|
week_end: Ending week number (inclusive)
|
|
|
|
Returns:
|
|
List of game dictionaries with team info
|
|
|
|
Example:
|
|
games = await client.get_schedule_games(season=13, week_start=12, week_end=12)
|
|
for game in games:
|
|
print(f"{game['away_abbrev']} @ {game['home_abbrev']}")
|
|
"""
|
|
url = f"{self.base_url}/games"
|
|
params = {"season": season, "week_start": week_start, "week_end": week_end}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
url, headers=self._get_headers(), params=params
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
games = data.get("games", data) if isinstance(data, dict) else data
|
|
|
|
logger.info(
|
|
f"Loaded {len(games)} scheduled games for S{season} W{week_start}-{week_end}"
|
|
)
|
|
return games
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Failed to fetch schedule games: {e}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error fetching schedule games: {e}")
|
|
raise
|
|
|
|
|
|
# Singleton instance
|
|
sba_api_client = SbaApiClient()
|