from sqlalchemy import Column, ForeignKey, Index, Integer, String, and_ from sqlalchemy.dialects.postgresql import ARRAY, JSONB from sqlalchemy.ext.declarative import declared_attr from sqlalchemy.future import select from sqlalchemy.orm import relationship, selectinload from sqlalchemy_json import mutable_json_type from server import config from server.db.models.base import Base, HiddenDeleteMixin class Application(HiddenDeleteMixin, Base): """Application model. Specific application (or project) that is supported by the service. Attributes: slug: unique application slug. name: application name. """ __tablename__ = "application" id = Column(Integer, primary_key=True, autoincrement=True) slug = Column(String(30), nullable=False, unique=True, index=True) name = Column(String(100), nullable=True) settings = Column(mutable_json_type(dbtype=JSONB, nested=True), nullable=True) types = Column(ARRAY(String(30)), nullable=True) class AppRelatedMixin: @declared_attr def app_slug(cls): return Column(String(30), ForeignKey(Application.slug), nullable=False) class Account(HiddenDeleteMixin, AppRelatedMixin, Base): """Account model. Account of a specific user in a specific application, store information specific to this pair. """ get_or_activate = config.ALLOW_REACTIVATE_ACCOUNTS __tablename__ = "account" __table_args__ = (Index("ix_app_slug_active_user", "app_slug", "user_id", "is_active"),) id = Column(Integer, primary_key=True, autoincrement=True) user_id = Column(String(50), nullable=False) devices = relationship("Device", back_populates="account") settings = relationship("Settings", back_populates="account") favorites = relationship("Favorites", back_populates="account") class AccountRelatedMixin: __mapper_args__ = {"eager_defaults": True} @declared_attr def account_id(cls): return Column(Integer, ForeignKey(Account.id), nullable=False) @declared_attr def account(cls): return relationship("Account") @classmethod def q_select_with_account(cls): q = select(cls).options(selectinload(cls.account)).join_from(cls, Account).where(Account.is_active.is_(True)) return q @classmethod def q_account_id(cls, app_slug: str, user_id: str): return Account.q_filter(select(Account.id), app_slug=app_slug, user_id=user_id).subquery() @classmethod def q_filter_by_account(cls, app_slug: str, user_id: str, query=None): if query is None: query = select(cls) q = query.join_from(cls, Account).where( and_(Account.app_slug == app_slug, Account.user_id == user_id, Account.is_active.is_(True)) ) return q class ApplicationSettingsHistory(Base): """Application Settings History model. Specific settings version(s) that is supported by the service. Attributes: slug: unique application slug. """ __tablename__ = "application_settings_history" id = Column(Integer, primary_key=True, autoincrement=True) slug = Column(String(30), nullable=False) version = Column(Integer, nullable=False) settings = Column(mutable_json_type(dbtype=JSONB, nested=True)) types = Column(ARRAY(String(30)))