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.
 
 
 
 
 

125 lines
4.1 KiB

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 '<User %r>' % 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}: <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')
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')