from sqlalchemy import text, func, select from sqlalchemy.orm import column_property from atlas_um.pgdb import pgdb from atlas_um.pgdb.internal_user_domain import InternalUserDomain class Auth0AccountRole(pgdb.Model): __tablename__ = "auth0_account_role" id = pgdb.Column(pgdb.Integer, primary_key=True) auth0_account_id = pgdb.Column( pgdb.Integer, pgdb.ForeignKey("auth0_account.id"), nullable=False ) role_id = pgdb.Column( pgdb.String, pgdb.ForeignKey("auth0_role.id"), nullable=False ) class Auth0Role(pgdb.Model): __tablename__ = "auth0_role" id = pgdb.Column(pgdb.String, primary_key=True) name = pgdb.Column(pgdb.String, nullable=False) description = pgdb.Column(pgdb.String, nullable=False) accounts = pgdb.relationship( "Auth0Account", secondary=Auth0AccountRole.__table__, back_populates="roles", ) class Auth0Account(pgdb.Model): """Local copy of the Auth0 user.""" __tablename__ = "auth0_account" id = pgdb.Column(pgdb.Integer(), primary_key=True) user_id = pgdb.Column(pgdb.String(), nullable=False) email = pgdb.Column(pgdb.String(), nullable=False, default="") email_verified = pgdb.Column(pgdb.Boolean(), nullable=False, default=False) username = pgdb.Column(pgdb.String(), nullable=False, default="") phone_number = pgdb.Column(pgdb.String(), nullable=False, default="") phone_verified = pgdb.Column(pgdb.Boolean, nullable=False, default=False) created_at = pgdb.Column( pgdb.DateTime(), nullable=False, server_default=text("TIMEZONE('utc', CURRENT_TIMESTAMP)"), ) updated_at = pgdb.Column( pgdb.DateTime(), nullable=False, server_default=text("TIMEZONE('utc', CURRENT_TIMESTAMP)"), ) identities = pgdb.Column(pgdb.JSON()) app_metadata = pgdb.Column(pgdb.JSON()) user_metadata = pgdb.Column(pgdb.JSON()) picture = pgdb.Column(pgdb.String(), nullable=False, default="") name = pgdb.Column(pgdb.String(), nullable=False, default="") nickname = pgdb.Column(pgdb.String(), nullable=False, default="") multifactor = pgdb.Column(pgdb.JSON()) last_ip = pgdb.Column(pgdb.String(), nullable=False, default="") last_login = pgdb.Column(pgdb.DateTime(), nullable=True) logins_count = pgdb.Column(pgdb.Integer(), nullable=False, default=0) blocked = pgdb.Column(pgdb.Boolean, nullable=False, default=False) given_name = pgdb.Column(pgdb.String(), nullable=False, default="") family_name = pgdb.Column(pgdb.String(), nullable=False, default="") is_external = column_property( func.split_part(email, "@", 2).notin_( select([InternalUserDomain.domain]).select_from(InternalUserDomain) ) ) external_states = pgdb.relationship( "Auth0AccountExternalState", back_populates="auth0_account" ) roles = pgdb.relationship( "Auth0Role", secondary=Auth0AccountRole.__table__, back_populates="accounts", ) def __repr__(self): return f"" def __str__(self): return self.user_id class Auth0AccountExternalState(pgdb.Model): __tablename__ = "auth0_account_external_state" id = pgdb.Column(pgdb.Integer, primary_key=True) auth0_account_id = pgdb.Column( pgdb.Integer, pgdb.ForeignKey("auth0_account.id"), nullable=False ) source = pgdb.Column(pgdb.String, nullable=False) is_active = pgdb.Column(pgdb.Boolean, default=False, nullable=False) last_login = pgdb.Column(pgdb.DateTime, nullable=True) # Relations auth0_account = pgdb.relationship( "Auth0Account", back_populates="external_states", )