7d397a2e22
UNSET sentinel pattern: - Add UNSET sentinel in protocols.py for nullable field updates - Fix inability to clear deck description (UNSET=keep, None=clear) - Fix repository inability to clear validation_errors Starter deck improvements: - Remove unused has_starter_deck from CollectionService - Add deprecation notes to old starter deck methods Validation improvements: - Add energy type validation in deck_validator.py - Add energy type validation in deck schemas - Add VALID_ENERGY_TYPES constant Game loading fix: - Fix get_deck_for_game silently skipping invalid cards - Now raises ValueError with clear error message Tests: - Add TestEnergyTypeValidation test class - Add TestGetDeckForGame test class - Add tests for validate_energy_types utility function Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
249 lines
7.5 KiB
Python
249 lines
7.5 KiB
Python
"""PostgreSQL implementation of DeckRepository.
|
|
|
|
This module provides the PostgreSQL-specific implementation of the
|
|
DeckRepository protocol using SQLAlchemy async sessions.
|
|
|
|
Example:
|
|
async with get_db_session() as db:
|
|
repo = PostgresDeckRepository(db)
|
|
decks = await repo.get_by_user(user_id)
|
|
"""
|
|
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models.deck import Deck
|
|
from app.repositories.protocols import UNSET, DeckEntry
|
|
|
|
|
|
def _to_dto(model: Deck) -> DeckEntry:
|
|
"""Convert ORM model to DTO."""
|
|
return DeckEntry(
|
|
id=model.id,
|
|
user_id=model.user_id,
|
|
name=model.name,
|
|
cards=model.cards or {},
|
|
energy_cards=model.energy_cards or {},
|
|
is_valid=model.is_valid,
|
|
validation_errors=model.validation_errors,
|
|
is_starter=model.is_starter,
|
|
starter_type=model.starter_type,
|
|
description=model.description,
|
|
created_at=model.created_at,
|
|
updated_at=model.updated_at,
|
|
)
|
|
|
|
|
|
class PostgresDeckRepository:
|
|
"""PostgreSQL implementation of DeckRepository.
|
|
|
|
Uses SQLAlchemy async sessions for database access.
|
|
|
|
Attributes:
|
|
_db: The async database session.
|
|
"""
|
|
|
|
def __init__(self, db: AsyncSession) -> None:
|
|
"""Initialize with database session.
|
|
|
|
Args:
|
|
db: SQLAlchemy async session.
|
|
"""
|
|
self._db = db
|
|
|
|
async def get_by_id(self, deck_id: UUID) -> DeckEntry | None:
|
|
"""Get a deck by its ID.
|
|
|
|
Args:
|
|
deck_id: The deck's UUID.
|
|
|
|
Returns:
|
|
DeckEntry if found, None otherwise.
|
|
"""
|
|
result = await self._db.execute(select(Deck).where(Deck.id == deck_id))
|
|
model = result.scalar_one_or_none()
|
|
return _to_dto(model) if model else None
|
|
|
|
async def get_by_user(self, user_id: UUID) -> list[DeckEntry]:
|
|
"""Get all decks for a user.
|
|
|
|
Args:
|
|
user_id: The user's UUID.
|
|
|
|
Returns:
|
|
List of all user's decks, ordered by name.
|
|
"""
|
|
result = await self._db.execute(
|
|
select(Deck).where(Deck.user_id == user_id).order_by(Deck.name)
|
|
)
|
|
return [_to_dto(model) for model in result.scalars().all()]
|
|
|
|
async def get_user_deck(self, user_id: UUID, deck_id: UUID) -> DeckEntry | None:
|
|
"""Get a specific deck owned by a user.
|
|
|
|
Combines ownership check with retrieval.
|
|
|
|
Args:
|
|
user_id: The user's UUID.
|
|
deck_id: The deck's UUID.
|
|
|
|
Returns:
|
|
DeckEntry if found and owned by user, None otherwise.
|
|
"""
|
|
result = await self._db.execute(
|
|
select(Deck).where(
|
|
Deck.id == deck_id,
|
|
Deck.user_id == user_id,
|
|
)
|
|
)
|
|
model = result.scalar_one_or_none()
|
|
return _to_dto(model) if model else None
|
|
|
|
async def count_by_user(self, user_id: UUID) -> int:
|
|
"""Count how many decks a user has.
|
|
|
|
Args:
|
|
user_id: The user's UUID.
|
|
|
|
Returns:
|
|
Number of decks owned by user.
|
|
"""
|
|
result = await self._db.execute(select(func.count(Deck.id)).where(Deck.user_id == user_id))
|
|
return result.scalar_one()
|
|
|
|
async def create(
|
|
self,
|
|
user_id: UUID,
|
|
name: str,
|
|
cards: dict[str, int],
|
|
energy_cards: dict[str, int],
|
|
is_valid: bool,
|
|
validation_errors: list[str] | None,
|
|
is_starter: bool = False,
|
|
starter_type: str | None = None,
|
|
description: str | None = None,
|
|
) -> DeckEntry:
|
|
"""Create a new deck.
|
|
|
|
Args:
|
|
user_id: The user's UUID.
|
|
name: Display name for the deck.
|
|
cards: Card ID to quantity mapping.
|
|
energy_cards: Energy type to quantity mapping.
|
|
is_valid: Whether deck passes validation.
|
|
validation_errors: List of validation error messages.
|
|
is_starter: Whether this is a starter deck.
|
|
starter_type: Type of starter deck if applicable.
|
|
description: Optional deck description.
|
|
|
|
Returns:
|
|
The created DeckEntry.
|
|
"""
|
|
deck = Deck(
|
|
user_id=user_id,
|
|
name=name,
|
|
cards=cards,
|
|
energy_cards=energy_cards,
|
|
is_valid=is_valid,
|
|
validation_errors=validation_errors,
|
|
is_starter=is_starter,
|
|
starter_type=starter_type,
|
|
description=description,
|
|
)
|
|
self._db.add(deck)
|
|
await self._db.commit()
|
|
await self._db.refresh(deck)
|
|
return _to_dto(deck)
|
|
|
|
async def update(
|
|
self,
|
|
deck_id: UUID,
|
|
name: str | None = None,
|
|
cards: dict[str, int] | None = None,
|
|
energy_cards: dict[str, int] | None = None,
|
|
is_valid: bool | None = None,
|
|
validation_errors: list[str] | None = UNSET, # type: ignore[assignment]
|
|
description: str | None = UNSET, # type: ignore[assignment]
|
|
) -> DeckEntry | None:
|
|
"""Update an existing deck.
|
|
|
|
Only provided (non-UNSET) fields are updated.
|
|
Use UNSET (default) to keep existing value, or None to clear.
|
|
|
|
Args:
|
|
deck_id: The deck's UUID.
|
|
name: New name (optional, None keeps existing).
|
|
cards: New card composition (optional, None keeps existing).
|
|
energy_cards: New energy composition (optional, None keeps existing).
|
|
is_valid: New validation status (optional, None keeps existing).
|
|
validation_errors: New errors (UNSET=keep, None=clear, list=set).
|
|
description: New description (UNSET=keep, None=clear, str=set).
|
|
|
|
Returns:
|
|
Updated DeckEntry, or None if deck not found.
|
|
"""
|
|
result = await self._db.execute(select(Deck).where(Deck.id == deck_id))
|
|
deck = result.scalar_one_or_none()
|
|
|
|
if deck is None:
|
|
return None
|
|
|
|
if name is not None:
|
|
deck.name = name
|
|
if cards is not None:
|
|
deck.cards = cards
|
|
if energy_cards is not None:
|
|
deck.energy_cards = energy_cards
|
|
if is_valid is not None:
|
|
deck.is_valid = is_valid
|
|
# Use UNSET pattern for nullable fields that can be cleared
|
|
if validation_errors is not UNSET:
|
|
deck.validation_errors = validation_errors
|
|
if description is not UNSET:
|
|
deck.description = description
|
|
|
|
await self._db.commit()
|
|
await self._db.refresh(deck)
|
|
return _to_dto(deck)
|
|
|
|
async def delete(self, deck_id: UUID) -> bool:
|
|
"""Delete a deck.
|
|
|
|
Args:
|
|
deck_id: The deck's UUID.
|
|
|
|
Returns:
|
|
True if deleted, False if not found.
|
|
"""
|
|
result = await self._db.execute(select(Deck).where(Deck.id == deck_id))
|
|
deck = result.scalar_one_or_none()
|
|
|
|
if deck is None:
|
|
return False
|
|
|
|
await self._db.delete(deck)
|
|
await self._db.commit()
|
|
return True
|
|
|
|
async def has_starter(self, user_id: UUID) -> tuple[bool, str | None]:
|
|
"""Check if user has a starter deck.
|
|
|
|
Args:
|
|
user_id: The user's UUID.
|
|
|
|
Returns:
|
|
Tuple of (has_starter, starter_type).
|
|
"""
|
|
result = await self._db.execute(
|
|
select(Deck.starter_type)
|
|
.where(
|
|
Deck.user_id == user_id,
|
|
Deck.is_starter == True, # noqa: E712
|
|
)
|
|
.limit(1)
|
|
)
|
|
starter_type = result.scalar_one_or_none()
|
|
return (starter_type is not None, starter_type)
|