You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

195 lines
6.0 KiB

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
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
def choices(enum) -> List[Tuple[str, str]]:
out = []
for item in enum:
out.append((, item.friendly_name()))
return out
def friendly_name(self) -> str:
return str(self)
def coerce(enum, name):
if isinstance(name, enum):
return name
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 '<User %r>' % self.username
def print(self):
return self.username + (" (org)" if 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)
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(''), 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) -> '':
return[self.game_mode](self.teams_count, self.get_configuration())
def lock(self) -> 'Game':
ses = get_session()
return ses.query(Game).filter_by(game_id=self.game_id).with_for_update().first()
def print(self) -> str:
name =
if not name:
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(''), 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 ==')
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(''), 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(''), nullable=True)
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')