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