Source code for luna_configurator.db.models

from sqlalchemy import Column, String, Integer, TIMESTAMP, Sequence, ForeignKey, Boolean
from sqlalchemy.dialects.oracle import CLOB
from sqlalchemy.ext.declarative import declarative_base

from crutches_on_wheels.utils.db_functions import currentDBTimestamp
from utils.enums import DBTypes
from . import metadata
from .models_config import DBConfig

Base = declarative_base()

DB_CURRENT_TIMESTAMP = currentDBTimestamp(DBConfig.dbType) if DBConfig.dbType else None


[docs]def findValueColumnType() -> (String, CLOB): """ Find value column type according to DB type Returns: type for value column """ if DBConfig.dbType == DBTypes.postgres.value: return String elif DBConfig.dbType == DBTypes.oracle.value: return CLOB
[docs]class Limitation(Base): """ Limitation table """ __tablename__ = 'limitation' __table_args__ = {'comment': 'Table with limitations.'} Base.metadata = metadata # setting name limitation_name = Column(String(128), primary_key=True, comment='limitation name') # Validation schema validation_schema = Column(findValueColumnType(), comment='validation schema') # Services for setting services = Column(findValueColumnType(), comment='list of services') # Default value default_value = Column(findValueColumnType(), comment='default limitation value') # description description = Column(String(128), default='', comment='limitation description') def __repr__(self): return f"<Limitation (name={self.limitation_name})>"
[docs]class Setting(Base): """ Setting table """ __tablename__ = 'setting' __table_args__ = {'comment': 'Table of settings.'} Base.metadata = metadata # setting id id = Column(Integer, Sequence('setting_id', start=1), primary_key=True, comment='setting id') # setting value value = Column(findValueColumnType(), comment='setting value') # description description = Column(String(128), comment='settings description') # create time create_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment='setting create time') # last update time last_update_time = Column(TIMESTAMP, server_default=DB_CURRENT_TIMESTAMP, comment='setting last update time') # setting name name = Column(String(128), ForeignKey(Limitation.limitation_name, ondelete='CASCADE'), comment='setting name') # whether a setting is a default setting is_default = Column(Boolean, nullable=False, default=False, comment='whether a setting is a default setting') def __repr__(self): return f"<Setting(id={self.id})>"
[docs]class Tag(Base): """ Tag table. """ __tablename__ = 'tag' __table_args__ = {'comment': 'Table of tags.'} Base.metadata = metadata # setting id id = Column(Integer, ForeignKey(Setting.id, ondelete='CASCADE'), comment='setting id') # setting name name = Column(String(128), primary_key=True, comment='setting name') # string with tag tag = Column(String(128), primary_key=True, comment='setting tag') def __repr__(self): return f"<Tag(id={self.id})>"
[docs]class Group(Base): """ Group table """ __tablename__ = 'group' __table_args__ = {'comment': 'Table of groups'} Base.metadata = metadata group_name = Column(String(128), primary_key=True, comment='Group name') description = Column(String(256), comment='Group description')
[docs]class GroupLimitation(Base): """ Limitation group link table """ __tablename__ = 'group_limitation' __table_args__ = {'comment': 'Table of limitation groups links'} Base.metadata = metadata group_name = Column(String(128), ForeignKey(Group.group_name, ondelete='CASCADE'), primary_key=True, comment='Group name') limitation_name = Column(String(128), ForeignKey(Limitation.limitation_name, ondelete='CASCADE'), primary_key=True, comment='limitation name')
[docs]class ConfigsMigration(Base): """ Configs migration """ __tablename__ = "configs_migrations" Base.metadata = metadata # settings migration revision revision = Column(String(36), primary_key=True, comment="settings migration revision")