strat-gameplay-webapp/backend/app/models/db_models.py
Cal Corum 874e24dc75 CLAUDE: Implement comprehensive dice roll system with persistence
Core Implementation:
- Created roll_types.py with AbRoll, JumpRoll, FieldingRoll, D20Roll dataclasses
- Implemented DiceSystem singleton with cryptographically secure random generation
- Added Roll model to db_models.py with JSONB storage for roll history
- Implemented save_rolls_batch() and get_rolls_for_game() in database operations

Testing:
- 27 unit tests for roll type dataclasses (100% passing)
- 35 unit tests for dice system (34/35 passing, 1 timing issue)
- 16 integration tests for database persistence (uses production DiceSystem)

Features:
- Unique roll IDs using secrets.token_hex()
- League-specific logic (SBA d100 rare plays, PD error-based rare plays)
- Automatic derived value calculation (d6_two_total, jump_total, error_total)
- Full audit trail with context metadata
- Support for batch saving rolls per inning

Technical Details:
- Fixed dataclass inheritance with kw_only=True for Python 3.13
- Roll data stored as JSONB for flexible querying
- Indexed on game_id, roll_type, league_id, team_id for efficient retrieval
- Supports filtering by roll type, team, and timestamp ordering

Note: Integration tests have async connection pool issue when run together
(tests work individually, fixture cleanup needed in follow-up branch)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-24 08:29:02 -05:00

322 lines
12 KiB
Python

from sqlalchemy import Column, Integer, String, Boolean, DateTime, JSON, Text, ForeignKey, Float, CheckConstraint, UniqueConstraint, func
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import UUID, JSONB
import uuid
import pendulum
from app.database.session import Base
class GameCardsetLink(Base):
"""Link table for PD games - tracks which cardsets are allowed"""
__tablename__ = "game_cardset_links"
game_id = Column(UUID(as_uuid=True), ForeignKey("games.id", ondelete="CASCADE"), primary_key=True)
cardset_id = Column(Integer, primary_key=True)
priority = Column(Integer, default=1, index=True)
# Relationships
game = relationship("Game", back_populates="cardset_links")
class RosterLink(Base):
"""Tracks eligible cards (PD) or players (SBA) for a game
PD League: Uses card_id to track which cards are rostered
SBA League: Uses player_id to track which players are rostered
Exactly one of card_id or player_id must be populated per row.
"""
__tablename__ = "roster_links"
# Surrogate primary key (allows nullable card_id/player_id)
id = Column(Integer, primary_key=True, autoincrement=True)
game_id = Column(UUID(as_uuid=True), ForeignKey("games.id", ondelete="CASCADE"), nullable=False, index=True)
card_id = Column(Integer, nullable=True) # PD only
player_id = Column(Integer, nullable=True) # SBA only
team_id = Column(Integer, nullable=False, index=True)
# Relationships
game = relationship("Game", back_populates="roster_links")
# Table-level constraints
__table_args__ = (
# Ensure exactly one ID is populated (XOR logic)
CheckConstraint(
'(card_id IS NOT NULL)::int + (player_id IS NOT NULL)::int = 1',
name='roster_link_one_id_required'
),
# Unique constraint for PD: one card per game
UniqueConstraint('game_id', 'card_id', name='uq_game_card'),
# Unique constraint for SBA: one player per game
UniqueConstraint('game_id', 'player_id', name='uq_game_player'),
)
class Game(Base):
"""Game model"""
__tablename__ = "games"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
league_id = Column(String(50), nullable=False, index=True)
home_team_id = Column(Integer, nullable=False)
away_team_id = Column(Integer, nullable=False)
status = Column(String(20), nullable=False, default="pending", index=True)
game_mode = Column(String(20), nullable=False)
visibility = Column(String(20), nullable=False)
current_inning = Column(Integer)
current_half = Column(String(10))
home_score = Column(Integer, default=0)
away_score = Column(Integer, default=0)
# AI opponent configuration
home_team_is_ai = Column(Boolean, default=False)
away_team_is_ai = Column(Boolean, default=False)
ai_difficulty = Column(String(20), nullable=True)
# Timestamps
created_at = Column(DateTime, default=lambda: pendulum.now('UTC').naive(), index=True)
started_at = Column(DateTime)
completed_at = Column(DateTime)
# Results
winner_team_id = Column(Integer)
game_metadata = Column(JSON, default=dict)
# Relationships
plays = relationship("Play", back_populates="game", cascade="all, delete-orphan")
lineups = relationship("Lineup", back_populates="game", cascade="all, delete-orphan")
cardset_links = relationship("GameCardsetLink", back_populates="game", cascade="all, delete-orphan")
roster_links = relationship("RosterLink", back_populates="game", cascade="all, delete-orphan")
session = relationship("GameSession", back_populates="game", uselist=False, cascade="all, delete-orphan")
rolls = relationship("Roll", back_populates="game", cascade="all, delete-orphan")
class Play(Base):
"""Play model - tracks individual plays/at-bats"""
__tablename__ = "plays"
id = Column(Integer, primary_key=True, autoincrement=True)
game_id = Column(UUID(as_uuid=True), ForeignKey("games.id", ondelete="CASCADE"), nullable=False, index=True)
play_number = Column(Integer, nullable=False)
# Game state at start of play
inning = Column(Integer, nullable=False, default=1)
half = Column(String(10), nullable=False)
outs_before = Column(Integer, nullable=False, default=0)
batting_order = Column(Integer, nullable=False, default=1)
away_score = Column(Integer, default=0)
home_score = Column(Integer, default=0)
# Players involved (ForeignKeys to Lineup)
batter_id = Column(Integer, ForeignKey("lineups.id"), nullable=False)
pitcher_id = Column(Integer, ForeignKey("lineups.id"), nullable=False)
catcher_id = Column(Integer, ForeignKey("lineups.id"), nullable=False)
defender_id = Column(Integer, ForeignKey("lineups.id"), nullable=True)
runner_id = Column(Integer, ForeignKey("lineups.id"), nullable=True)
# Base runners (ForeignKeys to Lineup for who's on base)
on_first_id = Column(Integer, ForeignKey("lineups.id"), nullable=True)
on_second_id = Column(Integer, ForeignKey("lineups.id"), nullable=True)
on_third_id = Column(Integer, ForeignKey("lineups.id"), nullable=True)
# Runner final positions (None = out, 1-4 = base)
on_first_final = Column(Integer, nullable=True)
on_second_final = Column(Integer, nullable=True)
on_third_final = Column(Integer, nullable=True)
batter_final = Column(Integer, nullable=True)
# Base state code for efficient queries (bit field: 1=1st, 2=2nd, 4=3rd, 7=loaded)
on_base_code = Column(Integer, default=0)
# Strategic decisions
defensive_choices = Column(JSON, default=dict)
offensive_choices = Column(JSON, default=dict)
# Play result
dice_roll = Column(String(50))
hit_type = Column(String(50))
result_description = Column(Text)
outs_recorded = Column(Integer, nullable=False, default=0)
runs_scored = Column(Integer, default=0)
# Defensive details
check_pos = Column(String(10), nullable=True)
error = Column(Integer, default=0)
# Batting statistics
pa = Column(Integer, default=0)
ab = Column(Integer, default=0)
hit = Column(Integer, default=0)
double = Column(Integer, default=0)
triple = Column(Integer, default=0)
homerun = Column(Integer, default=0)
bb = Column(Integer, default=0)
so = Column(Integer, default=0)
hbp = Column(Integer, default=0)
rbi = Column(Integer, default=0)
sac = Column(Integer, default=0)
ibb = Column(Integer, default=0)
gidp = Column(Integer, default=0)
# Baserunning statistics
sb = Column(Integer, default=0)
cs = Column(Integer, default=0)
# Pitching events
wild_pitch = Column(Integer, default=0)
passed_ball = Column(Integer, default=0)
pick_off = Column(Integer, default=0)
balk = Column(Integer, default=0)
# Ballpark power events
bphr = Column(Integer, default=0)
bpfo = Column(Integer, default=0)
bp1b = Column(Integer, default=0)
bplo = Column(Integer, default=0)
# Advanced analytics
wpa = Column(Float, default=0.0)
re24 = Column(Float, default=0.0)
# Earned/unearned runs
run = Column(Integer, default=0)
e_run = Column(Integer, default=0)
# Game situation flags
is_tied = Column(Boolean, default=False)
is_go_ahead = Column(Boolean, default=False)
is_new_inning = Column(Boolean, default=False)
in_pow = Column(Boolean, default=False)
# Play workflow
complete = Column(Boolean, default=False, index=True)
locked = Column(Boolean, default=False)
# Timestamps
created_at = Column(DateTime, default=lambda: pendulum.now('UTC').naive(), index=True)
# Extensibility (use for custom runner data like jump status, etc.)
play_metadata = Column(JSON, default=dict)
# Relationships
game = relationship("Game", back_populates="plays")
batter = relationship("Lineup", foreign_keys=[batter_id], lazy="joined")
pitcher = relationship("Lineup", foreign_keys=[pitcher_id], lazy="joined")
catcher = relationship("Lineup", foreign_keys=[catcher_id], lazy="joined")
defender = relationship("Lineup", foreign_keys=[defender_id])
runner = relationship("Lineup", foreign_keys=[runner_id])
on_first = relationship("Lineup", foreign_keys=[on_first_id])
on_second = relationship("Lineup", foreign_keys=[on_second_id])
on_third = relationship("Lineup", foreign_keys=[on_third_id])
@property
def ai_is_batting(self) -> bool:
"""Determine if current batting team is AI-controlled"""
# Type cast for Pylance - at runtime self.half is a string, not a Column
half_value: str = self.half # type: ignore[assignment]
if half_value == 'top':
return self.game.away_team_is_ai
else:
return self.game.home_team_is_ai
@property
def ai_is_fielding(self) -> bool:
"""Determine if current fielding team is AI-controlled"""
# Type cast for Pylance - at runtime self.half is a string, not a Column
half_value: str = self.half # type: ignore[assignment]
if half_value == 'top':
return self.game.home_team_is_ai
else:
return self.game.away_team_is_ai
class Lineup(Base):
"""Lineup model - tracks player assignments in a game
PD League: Uses card_id to track which cards are in the lineup
SBA League: Uses player_id to track which players are in the lineup
Exactly one of card_id or player_id must be populated per row.
"""
__tablename__ = "lineups"
id = Column(Integer, primary_key=True, autoincrement=True)
game_id = Column(UUID(as_uuid=True), ForeignKey("games.id", ondelete="CASCADE"), nullable=False, index=True)
team_id = Column(Integer, nullable=False, index=True)
# Polymorphic player reference
card_id = Column(Integer, nullable=True) # PD only
player_id = Column(Integer, nullable=True) # SBA only
position = Column(String(10), nullable=False)
batting_order = Column(Integer)
# Substitution tracking
is_starter = Column(Boolean, default=True)
is_active = Column(Boolean, default=True, index=True)
entered_inning = Column(Integer, default=1)
replacing_id = Column(Integer, nullable=True) # Lineup ID of player being replaced
after_play = Column(Integer, nullable=True) # Play number when substitution occurred
# Pitcher fatigue
is_fatigued = Column(Boolean, nullable=True)
# Extensibility
lineup_metadata = Column(JSON, default=dict)
# Relationships
game = relationship("Game", back_populates="lineups")
# Table-level constraints
__table_args__ = (
# Ensure exactly one ID is populated (XOR logic)
CheckConstraint(
'(card_id IS NOT NULL)::int + (player_id IS NOT NULL)::int = 1',
name='lineup_one_id_required'
),
)
class GameSession(Base):
"""Game session tracking - real-time WebSocket state"""
__tablename__ = "game_sessions"
game_id = Column(UUID(as_uuid=True), ForeignKey("games.id", ondelete="CASCADE"), primary_key=True)
connected_users = Column(JSON, default=dict)
last_action_at = Column(DateTime, default=lambda: pendulum.now('UTC').naive(), index=True)
state_snapshot = Column(JSON, default=dict)
# Relationships
game = relationship("Game", back_populates="session")
class Roll(Base):
"""
Stores dice roll history for auditing and analytics
Tracks all dice rolls with full context for game recovery and statistics.
Supports both SBA and PD leagues with polymorphic player_id.
"""
__tablename__ = "rolls"
roll_id = Column(String, primary_key=True)
game_id = Column(UUID(as_uuid=True), ForeignKey("games.id", ondelete="CASCADE"), nullable=False, index=True)
roll_type = Column(String, nullable=False, index=True) # 'ab', 'jump', 'fielding', 'd20'
league_id = Column(String, nullable=False, index=True)
# Auditing/Analytics fields
team_id = Column(Integer, index=True)
player_id = Column(Integer, index=True) # Polymorphic: Lineup.player_id (SBA) or Lineup.card_id (PD)
# Full roll data stored as JSONB for flexibility
roll_data = Column(JSONB, nullable=False) # Complete roll with all dice values
context = Column(JSONB) # Additional metadata (pitcher, inning, outs, etc.)
timestamp = Column(DateTime(timezone=True), nullable=False, index=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Relationships
game = relationship("Game", back_populates="rolls")