2a95316f04
- Add lifespan context manager to app/main.py with startup/shutdown hooks - Wire startup: init_db(), init_redis(), CardService.load_all() - Wire shutdown: close_db(), close_redis() - Add /health/ready endpoint for readiness checks - Add CORS middleware with configurable origins - Disable docs in production (only available in dev) - Export get_session_dependency from app/db/__init__.py for FastAPI DI - Add game_cache_ttl_seconds to Settings (configurable, was hardcoded) - Fix datetime.utcnow() deprecation (4 occurrences) -> datetime.now(UTC) - Update test to match S3 image URL (was placeholder CDN) All 974 tests passing.
339 lines
10 KiB
Python
339 lines
10 KiB
Python
"""Game models for Mantimon TCG.
|
|
|
|
This module defines models for game state management:
|
|
- ActiveGame: In-progress games (Postgres backup for Redis)
|
|
- GameHistory: Completed games with replay data
|
|
|
|
Example:
|
|
# Creating an active game
|
|
game = ActiveGame(
|
|
game_type=GameType.CAMPAIGN,
|
|
player1_id=user.id,
|
|
npc_id="grass_trainer_1",
|
|
rules_config={"prize_count": 4},
|
|
game_state=serialized_state
|
|
)
|
|
|
|
# Recording game history
|
|
history = GameHistory(
|
|
game_type=GameType.CAMPAIGN,
|
|
player1_id=user.id,
|
|
npc_id="grass_trainer_1",
|
|
winner_id=user.id,
|
|
end_reason=EndReason.PRIZES_TAKEN,
|
|
turn_count=15
|
|
)
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import DateTime, ForeignKey, Index, Integer, String
|
|
from sqlalchemy import Enum as SQLEnum
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from app.db.base import Base
|
|
|
|
if TYPE_CHECKING:
|
|
from app.db.models.user import User
|
|
|
|
|
|
class GameType(str, Enum):
|
|
"""Type of game being played.
|
|
|
|
Attributes:
|
|
CAMPAIGN: Single-player campaign match vs NPC.
|
|
FREEPLAY: Casual unranked match.
|
|
RANKED: Competitive ranked match.
|
|
PRACTICE: Practice mode (no rewards).
|
|
"""
|
|
|
|
CAMPAIGN = "campaign"
|
|
FREEPLAY = "freeplay"
|
|
RANKED = "ranked"
|
|
PRACTICE = "practice"
|
|
|
|
|
|
class EndReason(str, Enum):
|
|
"""Reason a game ended.
|
|
|
|
Attributes:
|
|
PRIZES_TAKEN: Winner took all prize cards.
|
|
NO_POKEMON: Opponent had no Pokemon left in play.
|
|
CANNOT_DRAW: Opponent couldn't draw at turn start.
|
|
RESIGNATION: Player resigned/quit.
|
|
TIMEOUT: Player timed out.
|
|
TURN_LIMIT: Game reached turn limit (draw possible).
|
|
DISCONNECTION: Player disconnected and didn't reconnect.
|
|
DRAW: Game ended in a draw.
|
|
"""
|
|
|
|
PRIZES_TAKEN = "prizes_taken"
|
|
NO_POKEMON = "no_pokemon"
|
|
CANNOT_DRAW = "cannot_draw"
|
|
RESIGNATION = "resignation"
|
|
TIMEOUT = "timeout"
|
|
TURN_LIMIT = "turn_limit"
|
|
DISCONNECTION = "disconnection"
|
|
DRAW = "draw"
|
|
|
|
|
|
class ActiveGame(Base):
|
|
"""In-progress game state (Postgres backup).
|
|
|
|
This is the durable storage for active games. The primary game state
|
|
lives in Redis for fast access during gameplay. Postgres serves as
|
|
backup and is updated at turn boundaries.
|
|
|
|
Attributes:
|
|
id: Unique game identifier (UUID).
|
|
game_type: Type of game (campaign, freeplay, ranked).
|
|
player1_id: First player's user ID.
|
|
player2_id: Second player's user ID (null for campaign).
|
|
npc_id: NPC opponent ID for campaign games.
|
|
rules_config: JSONB of RulesConfig settings.
|
|
game_state: JSONB of serialized GameState.
|
|
turn_number: Current turn number.
|
|
started_at: When the game started.
|
|
last_action_at: Timestamp of last action.
|
|
turn_deadline: When current turn times out.
|
|
|
|
Notes:
|
|
- Redis is primary, Postgres is backup
|
|
- Updated at turn boundaries (not every action)
|
|
- On server restart, active games load from Postgres -> Redis
|
|
"""
|
|
|
|
__tablename__ = "active_games"
|
|
|
|
# Game type
|
|
game_type: Mapped[GameType] = mapped_column(
|
|
SQLEnum(GameType, name="game_type", create_constraint=True),
|
|
nullable=False,
|
|
doc="Type of game (campaign, freeplay, ranked)",
|
|
)
|
|
|
|
# Players
|
|
player1_id: Mapped[UUID] = mapped_column(
|
|
ForeignKey("users.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
doc="First player's user ID",
|
|
)
|
|
player2_id: Mapped[UUID | None] = mapped_column(
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
doc="Second player's user ID (null for campaign)",
|
|
)
|
|
npc_id: Mapped[str | None] = mapped_column(
|
|
String(100),
|
|
nullable=True,
|
|
doc="NPC opponent ID for campaign games",
|
|
)
|
|
|
|
# Game configuration and state (JSONB)
|
|
rules_config: Mapped[dict] = mapped_column(
|
|
JSONB,
|
|
default=dict,
|
|
nullable=False,
|
|
doc="RulesConfig settings as JSONB",
|
|
)
|
|
game_state: Mapped[dict] = mapped_column(
|
|
JSONB,
|
|
default=dict,
|
|
nullable=False,
|
|
doc="Serialized GameState as JSONB",
|
|
)
|
|
turn_number: Mapped[int] = mapped_column(
|
|
Integer,
|
|
default=0,
|
|
nullable=False,
|
|
doc="Current turn number",
|
|
)
|
|
|
|
# Timing
|
|
started_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=lambda: datetime.now(UTC),
|
|
nullable=False,
|
|
doc="When the game started",
|
|
)
|
|
last_action_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=lambda: datetime.now(UTC),
|
|
nullable=False,
|
|
doc="Timestamp of last action",
|
|
)
|
|
turn_deadline: Mapped[datetime | None] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=True,
|
|
doc="When current turn times out",
|
|
)
|
|
|
|
# Relationships
|
|
player1: Mapped["User"] = relationship(
|
|
"User",
|
|
foreign_keys=[player1_id],
|
|
lazy="selectin",
|
|
)
|
|
player2: Mapped["User | None"] = relationship(
|
|
"User",
|
|
foreign_keys=[player2_id],
|
|
lazy="selectin",
|
|
)
|
|
|
|
# Indexes
|
|
__table_args__ = (
|
|
Index("ix_active_games_player1", "player1_id"),
|
|
Index("ix_active_games_player2", "player2_id"),
|
|
Index("ix_active_games_players", "player1_id", "player2_id"),
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<ActiveGame(id={self.id!r}, type={self.game_type}, turn={self.turn_number})>"
|
|
|
|
|
|
class GameHistory(Base):
|
|
"""Completed game record.
|
|
|
|
Stores the result of finished games including replay data
|
|
for future replay feature.
|
|
|
|
Attributes:
|
|
id: Unique identifier (UUID).
|
|
game_type: Type of game that was played.
|
|
player1_id: First player's user ID.
|
|
player2_id: Second player's user ID (null for campaign).
|
|
npc_id: NPC opponent ID for campaign games.
|
|
winner_id: User ID of the winner (null for draws).
|
|
winner_is_npc: True if NPC won (campaign games).
|
|
end_reason: How the game ended.
|
|
turn_count: Total turns played.
|
|
duration_seconds: Game duration in seconds.
|
|
replay_data: JSONB containing action log for replays.
|
|
played_at: When the game was completed.
|
|
|
|
Notes:
|
|
- Replay data captures all actions from game start
|
|
- Indexes optimize leaderboard and history queries
|
|
"""
|
|
|
|
__tablename__ = "game_history"
|
|
|
|
# Game type
|
|
game_type: Mapped[GameType] = mapped_column(
|
|
SQLEnum(GameType, name="game_type", create_constraint=True),
|
|
nullable=False,
|
|
doc="Type of game that was played",
|
|
)
|
|
|
|
# Players
|
|
# NOTE: nullable=True to allow SET NULL on cascade when user is deleted.
|
|
# Game history is preserved even after user deletion for leaderboards.
|
|
player1_id: Mapped[UUID | None] = mapped_column(
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
doc="First player's user ID (nullable for preserved history)",
|
|
)
|
|
player2_id: Mapped[UUID | None] = mapped_column(
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
doc="Second player's user ID",
|
|
)
|
|
npc_id: Mapped[str | None] = mapped_column(
|
|
String(100),
|
|
nullable=True,
|
|
doc="NPC opponent ID for campaign games",
|
|
)
|
|
|
|
# Result
|
|
winner_id: Mapped[UUID | None] = mapped_column(
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True,
|
|
doc="Winner's user ID (null for draws or NPC wins)",
|
|
)
|
|
winner_is_npc: Mapped[bool] = mapped_column(
|
|
default=False,
|
|
nullable=False,
|
|
doc="True if NPC won (campaign games)",
|
|
)
|
|
end_reason: Mapped[EndReason] = mapped_column(
|
|
SQLEnum(EndReason, name="end_reason", create_constraint=True),
|
|
nullable=False,
|
|
doc="How the game ended",
|
|
)
|
|
|
|
# Statistics
|
|
turn_count: Mapped[int] = mapped_column(
|
|
Integer,
|
|
nullable=False,
|
|
doc="Total turns played",
|
|
)
|
|
duration_seconds: Mapped[int] = mapped_column(
|
|
Integer,
|
|
nullable=False,
|
|
doc="Game duration in seconds",
|
|
)
|
|
|
|
# Replay data (captures all actions)
|
|
replay_data: Mapped[dict | None] = mapped_column(
|
|
JSONB,
|
|
nullable=True,
|
|
doc="Action log for replay feature",
|
|
)
|
|
|
|
# Timestamp
|
|
played_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=lambda: datetime.now(UTC),
|
|
nullable=False,
|
|
doc="When the game was completed",
|
|
)
|
|
|
|
# Relationships
|
|
# NOTE: passive_deletes=True lets database handle ON DELETE SET NULL
|
|
player1: Mapped["User | None"] = relationship(
|
|
"User",
|
|
foreign_keys=[player1_id],
|
|
lazy="selectin",
|
|
passive_deletes=True,
|
|
)
|
|
player2: Mapped["User | None"] = relationship(
|
|
"User",
|
|
foreign_keys=[player2_id],
|
|
lazy="selectin",
|
|
passive_deletes=True,
|
|
)
|
|
winner: Mapped["User | None"] = relationship(
|
|
"User",
|
|
foreign_keys=[winner_id],
|
|
lazy="selectin",
|
|
passive_deletes=True,
|
|
)
|
|
|
|
# Indexes for leaderboard and history queries
|
|
__table_args__ = (
|
|
Index("ix_game_history_player1", "player1_id"),
|
|
Index("ix_game_history_player2", "player2_id"),
|
|
Index("ix_game_history_winner", "winner_id"),
|
|
Index("ix_game_history_played_at", "played_at"),
|
|
Index("ix_game_history_type_played", "game_type", "played_at"),
|
|
)
|
|
|
|
@property
|
|
def is_draw(self) -> bool:
|
|
"""Check if game ended in a draw.
|
|
|
|
Returns:
|
|
True if no winner and end reason is DRAW.
|
|
"""
|
|
return self.end_reason == EndReason.DRAW
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<GameHistory(id={self.id!r}, type={self.game_type}, "
|
|
f"turns={self.turn_count}, end={self.end_reason})>"
|
|
)
|