from sqlalchemy import \ Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, \ text, func, \ create_engine, inspect, select from sqlalchemy.engine import Engine from sqlalchemy.orm import relationship, sessionmaker, Session, class_mapper, joinedload, aliased from sqlalchemy.orm.attributes import get_history from sqlalchemy.orm.query import Query from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql.expression import CTE from sqlalchemy.sql.functions import ReturnTypeFromArgs from sqlalchemy.sql.sqltypes import Numeric from typing import Any, Optional, List, Tuple import secrets import string import hra.config as config import hra.game Base = declarative_base() metadata = Base.metadata _engine: Optional[Engine] = None _session: Optional[Session] = None flask_db: Any = None def get_engine() -> Engine: global _engine if _engine is None: _engine = create_engine(config.SQLALCHEMY_DATABASE_URI, echo=config.SQLALCHEMY_ECHO) return _engine def get_session() -> Session: global _session if flask_db: return flask_db.session if _session is None: MOSession = sessionmaker(bind=get_engine()) _session = MOSession() return _session class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) org = Column(Boolean) token = Column(String(80), unique=True, nullable=False) username = Column(String(80), unique=True, nullable=False) passwd = Column(String(80), nullable=False) def gen_token(self): self.token = ''.join(secrets.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for i in range(15)) def __repr__(self): return '' % self.username def print(self): return self.username + (" (org)" if self.org else "") class Game(Base): __tablename__ = 'games' game_id = Column(Integer, primary_key=True) configuration = Column(JSONB, nullable=False) game_mode = Column(String(80), nullable=False) teams_count = Column(Integer, nullable=False) working_on_next_state = Column(Boolean, nullable=False, default=True) current_round = Column(Integer, default=-1) def current_state(self, none_if_working=True) -> Optional['State']: if none_if_working and self.working_on_next_state is True: return None return get_session().query(State).filter_by(game_id=self.game_id, round=self.current_round).order_by(State.round.desc()).first() def get_logic(self) -> 'hra.game.Logic': return hra.game.logic_by_mode[self.game_mode](self.teams_count, self.configuration) def lock(self) -> 'Game': ses = get_session() ses.expire_all() return ses.query(Game).filter_by(game_id=self.game_id).with_for_update().first() def print(self): return f"{self.game_id}: " class Team(Base): __tablename__ = 'base' __table_args__ = ( UniqueConstraint('game_id', 'team_id'), UniqueConstraint('user_id', 'name'), ) game_id = Column(Integer, ForeignKey('games.game_id'), nullable=False, primary_key=True) team_id = Column(Integer, nullable=False, primary_key=True) user_id = Column(Integer, ForeignKey('users.id'), nullable=True) name = Column(String(80), nullable=False) game = relationship('Game', primaryjoin='Team.game_id == Game.game_id') user = relationship('User', primaryjoin='Team.user_id == User.id') class State(Base): __tablename__ = 'states' game_id = Column(Integer, ForeignKey('games.game_id'), primary_key=True, nullable=False) round = Column(Integer, primary_key=True) state = Column(JSONB) game = relationship('Game', primaryjoin='State.game_id == Game.game_id') class Move(Base): __tablename__ = 'moves' game_id = Column(Integer, ForeignKey('games.game_id'), primary_key=True, nullable=False) round = Column(Integer, primary_key=True) team_id = Column(Integer, primary_key=True) move = Column(JSONB) game = relationship('Game', primaryjoin='Move.game_id == Game.game_id')