Files
bridzik/db/models.py
tim 30c32b7714 Add persistence layer: TOTP auth, game history, restore
- db/ package: async SQLAlchemy engine + Player/Game/Guess models
- api/auth.py: passwordless TOTP login (pyotp), session token via socket auth
- api/history.py: record guesses/points, DB-backed standings, restore
  unfinished games on startup, host-only end_game
- api/__init__.py: auth-gated handlers, accounts map, rejoin via account
- frontend: Auth (QR + code) and History pages, resume/end-game in lobby/table
- docker-compose: real PostgreSQL service wired via DATABASE_URL
- tests_history.py for the persistence/auth layer; refresh CLAUDE.md

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-23 23:09:50 +02:00

86 lines
3.1 KiB
Python

"""ORM modely: Player, Game, Guess.
Zamerne minimalne (3 tabulky). `won` sa neuklada -- vyplyva z `points > 0`
(trafeny tip = 10 + tip, inak 0; pozri Round.get_points_summary v bridzik.py).
"""
import uuid
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column
from db.db import Base
class Player(Base):
"""Trvaly ucet hraca + autentifikacia (TOTP)."""
__tablename__ = "players"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(String(40), unique=True, index=True)
totp_secret: Mapped[str] = mapped_column(String(32))
# Posledny pouzity TOTP casovy krok -- ochrana proti replay v ramci okna.
totp_last_step: Mapped[int] = mapped_column(Integer, default=0)
# Session token pre auto-reconnect (poslany v Socket.IO `auth`).
auth_token: Mapped[str | None] = mapped_column(
String(64), unique=True, nullable=True
)
created_at: Mapped[datetime] = mapped_column(
DateTime, server_default=func.now()
)
class Game(Base):
"""Jedna partia. Drzi priamo 4 ID hracov podla sedadla (0-3)."""
__tablename__ = "games"
id: Mapped[str] = mapped_column(
String(36), primary_key=True, default=lambda: str(uuid.uuid4())
)
name: Mapped[str] = mapped_column(String(40), default="")
player0_id: Mapped[int] = mapped_column(ForeignKey("players.id"))
player1_id: Mapped[int] = mapped_column(ForeignKey("players.id"))
player2_id: Mapped[int] = mapped_column(ForeignKey("players.id"))
player3_id: Mapped[int] = mapped_column(ForeignKey("players.id"))
# Aktualna pozicia hry. Spolu s rozdanim kariet nanovo plne urcuju stav hry
# po zaciatok kola -> staci na restore (pozri api/history.rebuild_core).
series: Mapped[int] = mapped_column(Integer, default=0)
round: Mapped[int] = mapped_column(Integer, default=0)
created_at: Mapped[datetime] = mapped_column(
DateTime, server_default=func.now()
)
ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
def player_id_for_seat(self, seat: int) -> int:
return (self.player0_id, self.player1_id, self.player2_id, self.player3_id)[seat]
class Guess(Base):
"""Tip a vysledok jedneho hraca v jednom kole.
Unikat (game_id, series_number, round_number, player_id) robi zapis
idempotentnym -- opakovany `record_completed_rounds` nezaklada duplikaty.
"""
__tablename__ = "guesses"
__table_args__ = (
UniqueConstraint(
"game_id",
"series_number",
"round_number",
"player_id",
name="uq_guess_round_player",
),
)
id: Mapped[int] = mapped_column(primary_key=True)
game_id: Mapped[str] = mapped_column(ForeignKey("games.id"), index=True)
player_id: Mapped[int] = mapped_column(ForeignKey("players.id"), index=True)
series_number: Mapped[int] = mapped_column(Integer)
round_number: Mapped[int] = mapped_column(Integer)
guess: Mapped[int] = mapped_column(Integer)
points: Mapped[int] = mapped_column(Integer)