strat-gameplay-webapp/backend/app/database/CLAUDE.md

Database Module - Async PostgreSQL Persistence

Purpose

Async PostgreSQL persistence layer using SQLAlchemy 2.0. Handles all database operations with connection pooling and proper transaction management.

Schema Management: Alembic migrations (see backend/README.md for commands)

Structure

app/database/
├── __init__.py      # Package exports
├── session.py       # Async session management, Base declarative
└── operations.py    # DatabaseOperations class (session injection pattern)

backend/alembic/     # Migration scripts (managed by Alembic)
├── env.py           # Migration environment config
└── versions/        # Migration files (001, 004, 005, etc.)

Session Injection Pattern (IMPORTANT)

DatabaseOperations supports session injection to enable:

• Transaction grouping: Multiple operations in one atomic transaction
• Test isolation: Tests inject sessions with automatic rollback
• Connection efficiency: Avoids asyncpg "operation in progress" conflicts

Two Usage Modes

Mode 1: Standalone (Default) - Each operation auto-commits

from app.database.operations import DatabaseOperations

db_ops = DatabaseOperations()  # No session injected
await db_ops.create_game(...)  # Creates session, commits, closes
await db_ops.save_play(...)    # Creates NEW session, commits, closes

Mode 2: Session Injection - Caller controls transaction

from app.database.operations import DatabaseOperations
from app.database.session import AsyncSessionLocal

async with AsyncSessionLocal() as session:
    try:
        db_ops = DatabaseOperations(session)  # Inject session
        await db_ops.save_play(...)           # Uses injected session (flush only)
        await db_ops.update_game_state(...)   # Same session (flush only)
        await session.commit()                # Caller commits once
    except Exception:
        await session.rollback()
        raise

When to Use Each Mode

Scenario	Mode	Why
Single operation	Standalone	Simpler, auto-commits
Multiple related operations	Session injection	Atomic transaction
Integration tests	Session injection	Rollback after test
game_engine.py transactions	Session injection	save_play + update_game_state atomic

Internal Implementation

class DatabaseOperations:
    def __init__(self, session: AsyncSession | None = None):
        self._session = session

    @asynccontextmanager
    async def _get_session(self):
        if self._session:
            yield self._session  # Caller controls commit/rollback
        else:
            async with AsyncSessionLocal() as session:
                try:
                    yield session
                    await session.commit()
                except Exception:
                    await session.rollback()
                    raise

Key Detail: Methods use flush() not commit() - this persists changes within the transaction but doesn't finalize it. For injected sessions, the caller commits. For standalone, the context manager commits.

DatabaseOperations API

Game Operations

from app.database.operations import DatabaseOperations
db_ops = DatabaseOperations()

await db_ops.create_game(game_id, league_id, home_team_id, away_team_id, ...)
await db_ops.update_game_state(game_id, inning, half, home_score, away_score)
game_data = await db_ops.load_game_state(game_id)

Play Operations

play_id = await db_ops.save_play(play_data)
plays = await db_ops.get_plays(game_id, limit=100)

Lineup Operations

lineup_id = await db_ops.add_pd_lineup_card(game_id, team_id, card_id, position, batting_order)
lineup_id = await db_ops.add_sba_lineup_player(game_id, team_id, player_id, position, batting_order)
lineup = await db_ops.get_active_lineup(game_id, team_id)
await db_ops.deactivate_lineup_player(lineup_id)

Roster Operations

await db_ops.add_pd_roster_card(game_id, card_id, team_id)
await db_ops.add_sba_roster_player(game_id, player_id, team_id)
roster = await db_ops.get_pd_roster(game_id, team_id)

Session Operations

await db_ops.create_game_session(game_id)
await db_ops.update_session_snapshot(game_id, state_snapshot)

Common Patterns

Transaction Grouping (Recommended for related operations)

async with AsyncSessionLocal() as session:
    try:
        db_ops = DatabaseOperations(session)
        await db_ops.save_play(play_data)
        await db_ops.update_game_state(game_id, ...)
        await session.commit()  # Both succeed or both fail
    except Exception:
        await session.rollback()
        raise

Efficient Queries

# Use joinedload for relationships needed immediately
from sqlalchemy.orm import joinedload

query = select(Play).options(
    joinedload(Play.batter),
    joinedload(Play.pitcher)
)

Direct UPDATE

# More efficient than SELECT + modify + commit
stmt = update(Game).where(Game.id == game_id).values(inning=5)
await session.execute(stmt)

Database Schema

See ../models/CLAUDE.md for model details.

Table	Purpose
games	Game container, scores, status
plays	At-bat records with 25+ stats
lineups	Player assignments, substitutions
game_sessions	WebSocket state
roster_links	Eligible cards/players

Environment

DATABASE_URL=postgresql+asyncpg://user:pass@10.10.0.42:5432/paperdynasty_dev

References

• Migrations: See backend/README.md for Alembic commands
• Models: See ../models/CLAUDE.md
• State Recovery: See ../core/state_manager.py

---

Tests: tests/integration/database/ (32 tests) | Updated: 2025-11-27