SKSP_2022_strategicka_hra/server/hra/db.py

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

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) -> Optional['State']:
        if 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':
        return get_session().query(Game).filter_by(game_id=self.game_id).with_for_update().first()


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')