#!/usr/bin/env python3 """ Major Domo API Client Shared API client for all Major Domo (SBA) operations. Provides methods for interacting with teams, players, standings, stats, transactions, and more. Environment Variables: API_TOKEN: Bearer token for API authentication (required) DATABASE: 'prod' or 'dev' (default: prod) """ import os import sys from typing import Optional, Dict, List, Any, Literal import requests class MajorDomoAPI: """ Major Domo API client for SBA database access Usage: api = MajorDomoAPI(environment='prod') # Get current season/week current = api.get_current() # Get a team team = api.get_team(abbrev='CAR') # List players players = api.list_players(season=12, team_id=42) # Get standings standings = api.get_standings(season=12, division_abbrev='ALE') """ def __init__(self, environment: str = 'prod', token: Optional[str] = None, verbose: bool = False): """ Initialize API client Args: environment: 'prod' or 'dev' token: API token (defaults to API_TOKEN env var) verbose: Print request/response details """ self.env = environment.lower() # Set base URL based on environment if 'prod' in self.env: self.base_url = 'https://api.sba.manticorum.com/v3' else: self.base_url = 'http://10.10.0.42:8000/api/v3' # Docker dev container self.token = token or os.getenv('API_TOKEN') self.verbose = verbose if not self.token: raise ValueError( "API_TOKEN environment variable required. " "Set it with: export API_TOKEN='your-token-here'" ) self.headers = { 'Authorization': f'Bearer {self.token}', 'Content-Type': 'application/json' } def _log(self, message: str): """Print message if verbose mode enabled""" if self.verbose: print(f"[API] {message}") def _build_url(self, endpoint: str, object_id: Optional[int] = None, **params) -> str: """Build API URL with query parameters""" url = f'{self.base_url}/{endpoint}' if object_id is not None: url += f'/{object_id}' # Add query parameters if params: param_parts = [] for key, value in params.items(): if value is not None: if isinstance(value, bool): param_parts.append(f'{key}={str(value).lower()}') elif isinstance(value, list): for item in value: param_parts.append(f'{key}={item}') else: param_parts.append(f'{key}={value}') if param_parts: url += '?' + '&'.join(param_parts) return url # ==================== # Low-level HTTP methods # ==================== def get(self, endpoint: str, object_id: Optional[int] = None, timeout: int = 10, **params) -> Any: """GET request to API""" url = self._build_url(endpoint, object_id=object_id, **params) self._log(f"GET {url}") response = requests.get(url, headers=self.headers, timeout=timeout) response.raise_for_status() return response.json() if response.text else None def post(self, endpoint: str, payload: Optional[Dict] = None, timeout: int = 10, **params) -> Any: """POST request to API""" url = self._build_url(endpoint, **params) self._log(f"POST {url}") response = requests.post(url, headers=self.headers, json=payload, timeout=timeout) response.raise_for_status() return response.json() if response.text else {} def patch(self, endpoint: str, object_id: Optional[int] = None, payload: Optional[Dict] = None, timeout: int = 10, **params) -> Any: """PATCH request to API""" url = self._build_url(endpoint, object_id=object_id, **params) self._log(f"PATCH {url}") response = requests.patch(url, headers=self.headers, json=payload, timeout=timeout) response.raise_for_status() return response.json() if response.text else {} def delete(self, endpoint: str, object_id: int, timeout: int = 10) -> str: """DELETE request to API""" url = self._build_url(endpoint, object_id=object_id) self._log(f"DELETE {url}") response = requests.delete(url, headers=self.headers, timeout=timeout) response.raise_for_status() return response.text # ==================== # Current Season/Week Operations # ==================== def get_current(self, season: Optional[int] = None) -> Dict: """ Get current season/week status Args: season: Specific season (defaults to latest) Returns: Current status dict with season, week, trade_deadline, etc. """ return self.get('current', season=season) def update_current(self, current_id: int, **updates) -> Dict: """ Update current season/week status (private endpoint) Args: current_id: Current record ID **updates: Fields to update (week, season, freeze, etc.) Returns: Updated current status dict """ return self.patch('current', object_id=current_id, **updates) # ==================== # Team Operations # ==================== def get_team(self, team_id: Optional[int] = None, abbrev: Optional[str] = None, season: Optional[int] = None) -> Dict: """ Get a team by ID or abbreviation Args: team_id: Team ID abbrev: Team abbreviation (e.g., 'CAR') season: Season number (if using abbrev) Returns: Team dict """ if team_id: return self.get('teams', object_id=team_id) elif abbrev: result = self.get('teams', team_abbrev=[abbrev.upper()], season=season) teams = result.get('teams', []) if not teams: raise ValueError(f"Team '{abbrev}' not found" + (f" in season {season}" if season else "")) return teams[0] else: raise ValueError("Must provide team_id or abbrev") def list_teams( self, season: Optional[int] = None, owner_id: Optional[List[int]] = None, manager_id: Optional[List[int]] = None, team_abbrev: Optional[List[str]] = None, active_only: bool = False, short_output: bool = False ) -> List[Dict]: """ List teams Args: season: Filter by season owner_id: Filter by owner Discord ID(s) manager_id: Filter by manager ID(s) team_abbrev: Filter by team abbreviation(s) active_only: Exclude IL and MiL teams short_output: Return minimal data Returns: List of team dicts """ result = self.get( 'teams', season=season, owner_id=owner_id, manager_id=manager_id, team_abbrev=team_abbrev, active_only=active_only, short_output=short_output ) return result.get('teams', []) def get_team_roster( self, team_id: int, which: Literal['current', 'next'] = 'current', sort: Optional[str] = None ) -> Dict: """ Get team roster breakdown Args: team_id: Team ID which: 'current' or 'next' week sort: Sort method (e.g., 'wara-desc') Returns: Roster dict with active/shortil/longil player lists """ return self.get(f'teams/{team_id}/roster/{which}', sort=sort) def update_team(self, team_id: int, **updates) -> Dict: """ Update team (private endpoint) Args: team_id: Team ID **updates: Fields to update (manager1_id, stadium, color, etc.) Returns: Updated team dict """ return self.patch('teams', object_id=team_id, **updates) # ==================== # Player Operations # ==================== def get_player( self, player_id: Optional[int] = None, name: Optional[str] = None, season: Optional[int] = None, short_output: bool = False ) -> Optional[Dict]: """ Get a player by ID or name Args: player_id: Player ID name: Player name (exact match) season: Season (required if using name) short_output: Return minimal data Returns: Player dict or None if not found """ if player_id: return self.get('players', object_id=player_id, short_output=short_output) elif name and season: result = self.get('players', season=season, name=name, short_output=short_output) players = result.get('players', []) return players[0] if players else None else: raise ValueError("Must provide player_id or (name and season)") def list_players( self, season: int, team_id: Optional[List[int]] = None, pos: Optional[List[str]] = None, strat_code: Optional[List[str]] = None, is_injured: Optional[bool] = None, sort: Optional[str] = None, short_output: bool = False ) -> List[Dict]: """ List players with filters Args: season: Season number (required) team_id: Filter by team ID(s) pos: Filter by position(s) (e.g., ['1B', 'OF']) strat_code: Filter by Strat code(s) is_injured: Only injured players sort: Sort method ('cost-asc', 'cost-desc', 'name-asc', 'name-desc') short_output: Return minimal data Returns: List of player dicts """ result = self.get( 'players', season=season, team_id=team_id, pos=pos, strat_code=strat_code, is_injured=is_injured, sort=sort, short_output=short_output ) return result.get('players', []) def search_players( self, query: str, season: Optional[int] = None, limit: int = 10, short_output: bool = False ) -> List[Dict]: """ Fuzzy search players by name Args: query: Search query (partial name match) season: Season (defaults to current) limit: Maximum results (1-50) short_output: Return minimal data Returns: List of player dicts """ result = self.get( 'players/search', q=query, season=season, limit=limit, short_output=short_output ) return result.get('players', []) def update_player(self, player_id: int, **updates) -> Dict: """ Update player (private endpoint) Args: player_id: Player ID **updates: Fields to update (team_id, wara, il_return, etc.) Returns: Updated player dict """ return self.patch('players', object_id=player_id, **updates) # ==================== # Standings Operations # ==================== def get_standings( self, season: int, team_id: Optional[List[int]] = None, league_abbrev: Optional[str] = None, division_abbrev: Optional[str] = None, short_output: bool = False ) -> List[Dict]: """ Get league standings Args: season: Season number team_id: Filter by team ID(s) league_abbrev: Filter by league (e.g., 'AL', 'NL') division_abbrev: Filter by division (e.g., 'ALE', 'NLW') short_output: Return minimal data Returns: List of standings dicts (sorted by win percentage) """ result = self.get( 'standings', season=season, team_id=team_id, league_abbrev=league_abbrev, division_abbrev=division_abbrev, short_output=short_output ) return result.get('standings', []) def get_team_standings(self, team_id: int) -> Dict: """ Get standings for a single team Args: team_id: Team ID Returns: Standings dict """ return self.get(f'standings/team/{team_id}') def recalculate_standings(self, season: int) -> str: """ Recalculate standings for a season (private endpoint) Args: season: Season number Returns: Success message """ return self.post(f'standings/s{season}/recalculate') # ==================== # Transaction Operations # ==================== def get_transactions( self, season: int, team_abbrev: Optional[List[str]] = None, week_start: Optional[int] = None, week_end: Optional[int] = None, cancelled: Optional[bool] = None, frozen: Optional[bool] = None, player_name: Optional[List[str]] = None, player_id: Optional[List[int]] = None, move_id: Optional[str] = None, short_output: bool = False ) -> List[Dict]: """ Get transactions with filters Args: season: Season number team_abbrev: Filter by team abbreviation(s) week_start: Start week (inclusive) week_end: End week (inclusive) cancelled: Filter by cancelled status frozen: Filter by frozen status player_name: Filter by player name(s) player_id: Filter by player ID(s) move_id: Filter by specific move ID short_output: Return minimal data Returns: List of transaction dicts """ result = self.get( 'transactions', season=season, team_abbrev=team_abbrev, week_start=week_start, week_end=week_end, cancelled=cancelled, frozen=frozen, player_name=player_name, player_id=player_id, move_id=move_id, short_output=short_output ) return result.get('transactions', []) # ==================== # Statistics Operations (Advanced Views) # ==================== def get_season_batting_stats( self, season: Optional[int] = None, team_id: Optional[int] = None, player_id: Optional[int] = None, sbaplayer_id: Optional[int] = None, min_pa: Optional[int] = None, sort_by: str = 'woba', sort_order: Literal['asc', 'desc'] = 'desc', limit: int = 200, offset: int = 0 ) -> List[Dict]: """ Get season batting statistics Args: season: Season number (defaults to current) team_id: Filter by team player_id: Filter by player sbaplayer_id: Filter by SBA player reference min_pa: Minimum plate appearances sort_by: Sort field (woba, avg, obp, slg, ops, hr, rbi, etc.) sort_order: 'asc' or 'desc' limit: Maximum results offset: Offset for pagination Returns: List of batting stat dicts """ result = self.get( 'views/season-stats/batting', season=season, team_id=team_id, player_id=player_id, sbaplayer_id=sbaplayer_id, min_pa=min_pa, sort_by=sort_by, sort_order=sort_order, limit=limit, offset=offset ) return result.get('stats', []) def get_season_pitching_stats( self, season: Optional[int] = None, team_id: Optional[int] = None, player_id: Optional[int] = None, sbaplayer_id: Optional[int] = None, min_outs: Optional[int] = None, sort_by: str = 'era', sort_order: Literal['asc', 'desc'] = 'asc', limit: int = 200, offset: int = 0 ) -> List[Dict]: """ Get season pitching statistics Args: season: Season number (defaults to current) team_id: Filter by team player_id: Filter by player sbaplayer_id: Filter by SBA player reference min_outs: Minimum outs pitched sort_by: Sort field (era, whip, k, bb, w, l, sv, etc.) sort_order: 'asc' or 'desc' limit: Maximum results offset: Offset for pagination Returns: List of pitching stat dicts """ result = self.get( 'views/season-stats/pitching', season=season, team_id=team_id, player_id=player_id, sbaplayer_id=sbaplayer_id, min_outs=min_outs, sort_by=sort_by, sort_order=sort_order, limit=limit, offset=offset ) return result.get('stats', []) def refresh_batting_stats(self, season: int) -> Dict: """ Refresh batting statistics for a season (private endpoint) Args: season: Season number Returns: Result dict with players_updated count """ return self.post(f'views/season-stats/batting/refresh', season=season) def refresh_pitching_stats(self, season: int) -> Dict: """ Refresh pitching statistics for a season (private endpoint) Args: season: Season number Returns: Result dict with players_updated count """ return self.post(f'views/season-stats/pitching/refresh', season=season) # ==================== # Schedule & Results Operations # ==================== def get_schedules( self, season: int, team_id: Optional[List[int]] = None, week: Optional[int] = None, short_output: bool = False ) -> List[Dict]: """ Get game schedules Args: season: Season number team_id: Filter by team ID(s) week: Filter by week short_output: Return minimal data Returns: List of schedule dicts """ result = self.get( 'schedules', season=season, team_id=team_id, week=week, short_output=short_output ) return result.get('schedules', []) def get_results( self, season: int, team_id: Optional[List[int]] = None, week: Optional[int] = None, short_output: bool = False ) -> List[Dict]: """ Get game results Args: season: Season number team_id: Filter by team ID(s) week: Filter by week short_output: Return minimal data Returns: List of result dicts """ result = self.get( 'results', season=season, team_id=team_id, week=week, short_output=short_output ) return result.get('results', []) # ==================== # Utility Methods # ==================== def health_check(self) -> bool: """ Check if API is accessible Returns: True if API is responding """ try: self.get_current() return True except Exception as e: if self.verbose: print(f"Health check failed: {e}") return False def __repr__(self) -> str: return f"MajorDomoAPI(env='{self.env}', base_url='{self.base_url}')" def main(): """CLI interface for testing""" import argparse parser = argparse.ArgumentParser(description='Major Domo API Client') parser.add_argument('--env', choices=['prod', 'dev'], default='prod', help='Environment') parser.add_argument('--verbose', action='store_true', help='Verbose output') args = parser.parse_args() try: api = MajorDomoAPI(environment=args.env, verbose=args.verbose) print(f"Connected to {api.env} environment") print(f"Base URL: {api.base_url}") # Test connection current = api.get_current() print(f"\nCurrent Status:") print(f" Season: {current['season']}") print(f" Week: {current['week']}") print(f" Freeze: {current['freeze']}") print(f" Trade Deadline: Week {current['trade_deadline']}") # List teams teams = api.list_teams(season=current['season'], active_only=True) print(f"\nActive Teams in Season {current['season']}: {len(teams)}") for team in teams[:5]: print(f" {team['abbrev']}: {team['lname']}") if len(teams) > 5: print(f" ... and {len(teams) - 5} more") print("\nāœ… API client working correctly!") except Exception as e: print(f"āŒ Error: {e}") sys.exit(1) if __name__ == '__main__': main()