#!/usr/bin/env python3 """ Interactive Database Playground Test and experiment with database models and operations. Run this script to create test games and explore the database operations. Usage: python test_db_playground.py """ import asyncio import logging from uuid import uuid4, UUID from typing import Optional from app.database.operations import DatabaseOperations from app.models.roster_models import PdRosterLinkData, SbaRosterLinkData # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class DatabasePlayground: """Interactive playground for testing database operations.""" def __init__(self): self.db_ops = DatabaseOperations() self.test_game_id: Optional[UUID] = None async def demo_pd_game(self): """Create and populate a PD league game with roster and lineup.""" logger.info("=" * 60) logger.info("DEMO: Creating PD League Game") logger.info("=" * 60) # Create game self.test_game_id = uuid4() game = await self.db_ops.create_game( game_id=self.test_game_id, league_id="pd", home_team_id=1, away_team_id=2, game_mode="friendly", visibility="public", home_team_is_ai=False, away_team_is_ai=True, ai_difficulty="balanced" ) logger.info(f"āœ… Created PD game: {game.id}") logger.info(f" Home Team: {game.home_team_id} (Human)") logger.info(f" Away Team: {game.away_team_id} (AI - {game.ai_difficulty})") # Add cards to roster logger.info("\nšŸ“‹ Adding cards to roster...") home_cards = [101, 102, 103, 104, 105, 106, 107, 108, 109] away_cards = [201, 202, 203, 204, 205, 206, 207, 208, 209] for card_id in home_cards: roster_link = await self.db_ops.add_pd_roster_card( game_id=self.test_game_id, card_id=card_id, team_id=1 ) logger.info(f" Added card {card_id} to home roster (link_id={roster_link.id})") for card_id in away_cards: roster_link = await self.db_ops.add_pd_roster_card( game_id=self.test_game_id, card_id=card_id, team_id=2 ) logger.info(f" Added card {card_id} to away roster (link_id={roster_link.id})") # Get roster home_roster = await self.db_ops.get_pd_roster(self.test_game_id, team_id=1) away_roster = await self.db_ops.get_pd_roster(self.test_game_id, team_id=2) logger.info(f"\nāœ… Home roster: {len(home_roster)} cards") logger.info(f"āœ… Away roster: {len(away_roster)} cards") # Set lineup logger.info("\n⚾ Setting starting lineup...") positions = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF'] for i, (card_id, position) in enumerate(zip(home_cards, positions), start=1): lineup = await self.db_ops.add_pd_lineup_card( game_id=self.test_game_id, team_id=1, card_id=card_id, position=position, batting_order=i if position != 'P' else None, is_starter=True ) logger.info(f" {position}: Card {card_id} (batting {lineup.batting_order or 'N/A'})") # Get active lineup home_lineup = await self.db_ops.get_active_lineup(self.test_game_id, team_id=1) logger.info(f"\nāœ… Active lineup: {len(home_lineup)} players") # Update game state logger.info("\nšŸŽ® Simulating game progress...") await self.db_ops.update_game_state( game_id=self.test_game_id, inning=3, half="bottom", home_score=2, away_score=1, status="active" ) logger.info(" Updated to: Bottom 3rd, Home 2 - Away 1") # Save a play play_data = { 'game_id': self.test_game_id, 'play_number': 1, 'inning': 3, 'half': 'bottom', 'outs_before': 1, 'batting_order': 3, 'away_score': 1, 'home_score': 2, 'batter_id': home_lineup[2].id, 'pitcher_id': home_lineup[0].id, 'catcher_id': home_lineup[1].id, 'dice_roll': '14+6', 'hit_type': 'FB', 'result_description': 'Deep fly ball to center field - Home Run!', 'outs_recorded': 0, 'runs_scored': 1, 'pa': 1, 'ab': 1, 'hit': 1, 'homerun': 1, 'rbi': 1, 'is_go_ahead': False, 'complete': True, 'locked': False } play = await self.db_ops.save_play(play_data) logger.info(f"\n⚾ Saved play #{play.play_number}: {play.result_description}") # Load game state logger.info("\nšŸ”„ Loading complete game state...") game_state = await self.db_ops.load_game_state(self.test_game_id) if game_state: logger.info(f"āœ… Loaded game state:") logger.info(f" Status: {game_state['game']['status']}") logger.info(f" Score: {game_state['game']['away_score']}-{game_state['game']['home_score']}") logger.info(f" Lineups: {len(game_state['lineups'])} players") logger.info(f" Plays: {len(game_state['plays'])} plays recorded") return self.test_game_id async def demo_sba_game(self): """Create and populate an SBA league game with roster and lineup.""" logger.info("\n" + "=" * 60) logger.info("DEMO: Creating SBA League Game") logger.info("=" * 60) # Create game game_id = uuid4() game = await self.db_ops.create_game( game_id=game_id, league_id="sba", home_team_id=10, away_team_id=11, game_mode="ranked", visibility="public", home_team_is_ai=False, away_team_is_ai=False ) logger.info(f"āœ… Created SBA game: {game.id}") logger.info(f" Home Team: {game.home_team_id}") logger.info(f" Away Team: {game.away_team_id}") logger.info(f" Mode: {game.game_mode}") # Add players to roster logger.info("\nšŸ“‹ Adding players to roster...") home_players = [501, 502, 503, 504, 505, 506, 507, 508, 509] away_players = [601, 602, 603, 604, 605, 606, 607, 608, 609] for player_id in home_players: roster_link = await self.db_ops.add_sba_roster_player( game_id=game_id, player_id=player_id, team_id=10 ) logger.info(f" Added player {player_id} to home roster (link_id={roster_link.id})") for player_id in away_players: roster_link = await self.db_ops.add_sba_roster_player( game_id=game_id, player_id=player_id, team_id=11 ) logger.info(f" Added player {player_id} to away roster (link_id={roster_link.id})") # Get roster home_roster = await self.db_ops.get_sba_roster(game_id, team_id=10) away_roster = await self.db_ops.get_sba_roster(game_id, team_id=11) logger.info(f"\nāœ… Home roster: {len(home_roster)} players") logger.info(f"āœ… Away roster: {len(away_roster)} players") # Set lineup logger.info("\n⚾ Setting starting lineup...") positions = ['P', 'C', '1B', '2B', '3B', 'SS', 'LF', 'CF', 'RF'] for i, (player_id, position) in enumerate(zip(home_players, positions), start=1): lineup = await self.db_ops.add_sba_lineup_player( game_id=game_id, team_id=10, player_id=player_id, position=position, batting_order=i if position != 'P' else None, is_starter=True ) logger.info(f" {position}: Player {player_id} (batting {lineup.batting_order or 'N/A'})") # Create game session session = await self.db_ops.create_game_session(game_id) logger.info(f"\nāœ… Created game session for game: {session.game_id}") # Update session snapshot await self.db_ops.update_session_snapshot( game_id=game_id, state_snapshot={'test': 'data', 'connected_users': ['user1', 'user2']} ) logger.info("āœ… Updated session snapshot") return game_id async def demo_cleanup(self): """Demonstrate querying capabilities.""" if not self.test_game_id: logger.warning("No test game created yet") return logger.info("\n" + "=" * 60) logger.info("DEMO: Query Capabilities") logger.info("=" * 60) # Get game game = await self.db_ops.get_game(self.test_game_id) if game: logger.info(f"\nšŸ“Š Game Details:") logger.info(f" ID: {game.id}") logger.info(f" League: {game.league_id}") logger.info(f" Status: {game.status}") logger.info(f" Inning: {game.current_half} {game.current_inning}") logger.info(f" Score: {game.away_score}-{game.home_score}") # Get plays plays = await self.db_ops.get_plays(self.test_game_id) logger.info(f"\n⚾ Plays: {len(plays)} total") for play in plays: logger.info(f" Play #{play.play_number}: {play.result_description}") # Get active lineup home_lineup = await self.db_ops.get_active_lineup(self.test_game_id, team_id=1) logger.info(f"\nšŸ‘„ Active Lineup: {len(home_lineup)} players") for player in sorted(home_lineup, key=lambda x: x.batting_order or 99): logger.info(f" {player.position}: Card {player.card_id} (batting {player.batting_order or 'N/A'})") async def main(): """Run interactive playground demos.""" playground = DatabasePlayground() try: # Run PD game demo pd_game_id = await playground.demo_pd_game() # Run SBA game demo sba_game_id = await playground.demo_sba_game() # Show query capabilities await playground.demo_cleanup() logger.info("\n" + "=" * 60) logger.info("āœ… ALL DEMOS COMPLETED SUCCESSFULLY!") logger.info("=" * 60) logger.info(f"\nCreated games:") logger.info(f" PD Game: {pd_game_id}") logger.info(f" SBA Game: {sba_game_id}") logger.info("\nYou can query these games in the database to explore the data!") except Exception as e: logger.error(f"\nāŒ Error during demo: {e}", exc_info=True) raise if __name__ == "__main__": asyncio.run(main())