association_object_ternary.py

"""
https://docs.sqlalchemy.org/en/14/orm/basic_relationships.html#association-object
"""
from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy import Column, ForeignKey, Integer, Text, DateTime
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy.orm.session import Session

# Mix ORM with Core SQL Expression language ???
# https://docs.sqlalchemy.org/en/14/core/tutorial.html
from sqlalchemy import select

Base = declarative_base()


class UserOrgRole(Base):
    __tablename__ = "user_org_role"

    user_id = Column(
        ForeignKey("user.id", ondelete="CASCADE"), primary_key=True
    )
    org_id = Column(ForeignKey("org.id", ondelete="CASCADE"), primary_key=True)
    # Does it make sense ?
    role_id = Column(
        ForeignKey("role.id", ondelete="CASCADE"), primary_key=True
    )

    # I do not know what the relationship could back populate, org or role ???
    user = relationship("User", back_populates="orgsroles")
    org = relationship("Org", back_populates="usersroles")
    role = relationship("Role", back_populates="usersorgs")

    def __repr__(self):
        return (
            f"<UserOrgRole (user_id={self.user_id}, org_id={self.org_id}, "
            f"role_id={self.role_id})>"
        )


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True)
    username = Column(Text, index=True, nullable=False)
    fullname = Column(Text, nullable=False)
    account_type = Column(Text, nullable=False)

    orgsroles = relationship("UserOrgRole", back_populates="user")

    def __repr__(self):
        return (
            f"<User (username={self.username}, fullname={self.fullname}, "
            f"account_type={self.account_type})>"
        )

    def get_roles(self, session):
        stmt = (
            select(Role)
            .select_from(UserOrgRole)
            .join(Role, UserOrgRole.role_id == Role.id)
            .where(UserOrgRole.user_id == self.id)
        )
        print(stmt)
        results = session.execute(stmt)
        print(results)
        # <sqlalchemy.engine.result.ChunkedIteratorResult object at 0x7f35d3bdd040>
        # https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.ChunkedIteratorResult
        for row in results:
            # Pourquoi row semble être un tuple qui contient l'objet ?
            # type(row)
            # <class 'sqlalchemy.engine.row.Row'>
            # type(row[0])
            # <class '__main__.Role'>
            print(row[0].name)

    def get_uor(self, session):
        # https://docs.sqlalchemy.org/en/14/orm/session_api.html?highlight=session%20execute#sqlalchemy.orm.Session.execute
        stmt = select(UserOrgRole).where(UserOrgRole.user_id == self.id)
        print(stmt)
        results = session.execute(stmt)
        for row in results:
            uor = row[0]
            print(uor)


class Org(Base):
    __tablename__ = "org"

    id = Column(Integer, primary_key=True)
    name = Column(Text, index=True, nullable=False)
    slug = Column(Text, index=True, nullable=False)
    created_at = Column(DateTime)

    usersroles = relationship("UserOrgRole", back_populates="org")

    def __repr__(self):
        return (
            f"<Org (name={self.name}, slug={self.slug}, "
            f"created_at={self.created_at})>"
        )


class Role(Base):
    __tablename__ = "role"

    id = Column(Integer, primary_key=True)
    name = Column(Text, index=True, nullable=False)

    usersorgs = relationship("UserOrgRole", back_populates="role")

    def __repr__(self):
        return f"<Role (name={self.name})>"


if __name__ == "__main__":
    engine = create_engine(
        "sqlite:///association_object_ternary.db", echo=False
    )

    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        # create parent, append a child via association
        u1 = User(
            username="jlondon",
            fullname="Jack London",
            account_type="member",
        )

        u2 = User(
            username="jdoe",
            fullname="John Doe",
            account_type="member",
        )

        o1 = Org(name="o1", slug="o1", created_at=datetime.utcnow())
        o2 = Org(name="o2", slug="o2", created_at=datetime.utcnow())

        owner = Role(name="owner")
        reader = Role(name="reader")

        uor1 = UserOrgRole()
        uor1.user = u1
        uor1.org = o1
        uor1.role = owner

        uor2 = UserOrgRole()
        uor2.user = u2
        uor2.org = o1
        uor2.role = reader

        uor3 = UserOrgRole()
        uor3.user = u2
        uor3.org = o2
        uor3.role = owner

        with session.begin():
            session.add(u1)
            session.add(u2)
            session.add(o1)
            session.add(o2)
            session.add(owner)
            session.add(reader)

            session.add(uor1)
            session.add(uor2)
            session.add(uor3)

        # u1.get_uor(session)
        u1.get_roles(session)

        print(u1)
        print(u1.orgsroles[0].org)
        print(u1.orgsroles[0].role)

        print(u2)
        print(u2.orgsroles[0].org)
        print(u2.orgsroles[0].role)

        print(o1)
        print(owner)

    """
    # Mix ORM with Core SQL Expression language ???
    user = Base.metadata.tables["user"]

    with engine.connect() as conn:
        stmt = select(user)
        for row in conn.execute(stmt):
            print(row)
    """

    # Y a-t-il des pistes dans les ORM examples ?
    # https://docs.sqlalchemy.org/en/14/_modules/examples/join_conditions/threeway.html

    # https://dba.stackexchange.com/questions/150865/how-to-model-a-three-way-association-that-involves-product-category-and-label