from sqlalchemy import \ Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, Enum, \ 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 from enum import Enum as PythonEnum, auto 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 class MOEnum(str, PythonEnum): """MOEnum je varianta PythonEnum, ve které se automaticky přidělované hodnoty jmenují stejně jako klíče a funguje serializace do JSONu.""" def _generate_next_value_(name, start, count, last_values): return name @classmethod def choices(enum) -> List[Tuple[str, str]]: out = [] for item in enum: out.append((item.name, item.friendly_name())) return out def friendly_name(self) -> str: return str(self) @classmethod def coerce(enum, name): if isinstance(name, enum): return name try: return enum[name] except KeyError: raise ValueError(name) 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 BigData(Base): __tablename__ = 'bigdata' id = Column(Integer, primary_key=True) data = Column(JSONB, nullable=False) def get_big_data(id): return get_session().query(BigData).filter_by(id=id).one_or_none().data def new_big_data(d): o = BigData(data=d) get_session().add(o) get_session().flush() return o.id class StepMode(MOEnum): none = auto() org = auto() user = auto() automatic = auto() class Game(Base): __tablename__ = 'games' game_id = Column(Integer, primary_key=True) name = Column(String(80), nullable=True) configuration = Column(Integer, ForeignKey('bigdata.id'), nullable=False) step_mode = Column(Enum(StepMode, name='step_mode'), nullable=False, default=StepMode.none) step_every_s = Column(Integer, nullable=False, default=60) 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 get_configuration(self): return get_big_data(self.configuration) 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).one_or_none() def get_logic(self) -> 'hra.game.Logic': return hra.game.logic_by_mode[self.game_mode](self.teams_count, self.get_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) -> str: name = self.name if not name: name = "" return f"{self.game_id}: {name}" 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') def print(self): if self.user is None: return f"{self.team_id}: -" return f"{self.team_id}: {self.user.username}" def get_move(self, round_id): return get_session().query(Move).filter_by(game_id=self.game_id, round=round_id, team_id=self.team_id).one_or_none() class State(Base): __tablename__ = 'states' create_time = Column(DateTime, nullable=False) game_id = Column(Integer, ForeignKey('games.game_id'), primary_key=True, nullable=False) round = Column(Integer, primary_key=True) state = Column(Integer, ForeignKey('bigdata.id'), nullable=False) def get_state(self): return get_big_data(self.state) 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(Integer, ForeignKey('bigdata.id'), nullable=True) points = Column(Integer, nullable=False, default=0) reads_count = Column(Integer, nullable=False, default=0) ok_pushs_count = Column(Integer, nullable=False, default=0) warnings_pushs_count = Column(Integer, nullable=False, default=0) err_pushs_count = Column(Integer, nullable=False, default=0) def get_move(self): if self.move is None: return None return get_big_data(self.move) game = relationship('Game', primaryjoin='Move.game_id == Game.game_id')