SKSP_2022_strategicka_hra/server/hra/db.py

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 '<User %r>' % 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 = "<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}"


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