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

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)


    def current_state(self) -> 'State':
        return get_session().query(State).filter_by(game_id=self.game_id).order_by(State.round.desc()).first()


class State(Base):
    __tablename__ = 'states'

    game_id = Column(Integer, ForeignKey('games.game_id'), primary_key=True, nullable=False)
    round = Column(Integer)
    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)
    team_id = Column(Integer)
    move = Column(JSONB)

    game = relationship('Game', primaryjoin='Move.game_id == Game.game_id')