SQLAlchemy 2.0 foreign key across parent and child tables

Question

Say I have these tables which define hierarchy

Org

• id
• name

SubOrg

• id
• org_id
• name

Now there is another table which refers both such that:

Rules

• id
• org_id
• sub_org_id (nullable)
• ...

The way Rules table works is if sub_org_id is null, the rule applies to whole org. if sub_ord_id is not null, that rule overrides for that particular sub_org.

How do I map this into sqlalchemy relationship?

Thanks

Answer 1

I would just use regular queries to resolve the rules. Maybe provide more info about the issue.

import os

from sqlalchemy import (
    Column,
    Integer,
    String,
    create_engine,
    ForeignKey,
)
from sqlalchemy.sql import (
    select,
    or_,
)
from sqlalchemy.orm import (
    declarative_base,
    Session,
    relationship
)


def get_engine(env):
    return create_engine(f"postgresql+psycopg2://{env['DB_USER']}:{env['DB_PASSWORD']}@{env['DB_HOST']}:{env['DB_PORT']}/{env['DB_NAME']}", echo=False)

Base = declarative_base()


class Org(Base):
    __tablename__ = 'orgs'

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

    all_rules = relationship("Rule",  back_populates="org")
    sub_orgs = relationship("SubOrg",  back_populates="org")

class SubOrg(Base):
    __tablename__ = 'sub_orgs'

    id = Column(Integer, primary_key=True)
    org_id = Column(Integer, ForeignKey("orgs.id"), nullable=False)
    name = Column(String, nullable=False)

    org = relationship("Org",  back_populates="sub_orgs")
    sub_org_only_rules = relationship("Rule",  back_populates="sub_org")


class Rule(Base):
    __tablename__ = 'rules'

    id = Column(Integer, primary_key=True)
    org_id = Column(Integer, ForeignKey("orgs.id"), nullable=False)
    sub_org_id = Column(Integer, ForeignKey("sub_orgs.id"), nullable=True)
    name = Column(String, nullable=False)

    org = relationship("Org",  back_populates="all_rules")
    sub_org = relationship("SubOrg",  back_populates="sub_org_only_rules")


def get_sub_org_rules(db, sub_org):
    q = select(Rule).where(Rule.org == sub_org.org, or_(Rule.sub_org == sub_org, Rule.sub_org_id == None)).order_by(Rule.sub_org_id.desc().nullslast(), Rule.name)
    return db.scalars(q).all()


def get_default_org_rules(db, org):
    q = select(Rule).where(Rule.org == org, Rule.sub_org_id == None).order_by(Rule.name)
    return db.scalars(q).all()


def query(conn):

    with Session(conn) as session:
        for org in session.scalars(select(Org).order_by(Org.name)):
            print (f"{org.name}({org.id})")
            print (f"{' '*4}Default Org Rules:")
            for rule in get_default_org_rules(session, org):
                print (f"{' '*6}#{rule.id} {rule.name}")
            print ('')
            for sub_org in org.sub_orgs:
                print (f"{' '*4}{sub_org.name}({sub_org.id})")
                print (f"{' '*8}Sub Org Rules:")
                for rule in get_sub_org_rules(session, sub_org):
                    print (f"{' '*10}#{rule.id} {rule.name}")
                print ('')

def create_sub_orgs(org, count):
    return [SubOrg(org=org, name=f"{org.name}-{i}") for i in range(count)]

def populate(conn):
    ORG_NAMES = ['A', 'B', 'C']
    orgs = {}
    with Session(conn) as session:
        for org_name in ORG_NAMES:
            orgs[org_name] = Org(name=org_name)
        session.add_all(orgs.values())

        session.add_all(create_sub_orgs(orgs['A'], 1))
        session.add_all(create_sub_orgs(orgs['C'], 2))
        session.commit()

        sub_orgs = dict([(sub_org.name, sub_org) for sub_org in session.scalars(select(SubOrg).join(SubOrg.org).order_by(Org.name, SubOrg.name))])

        session.add(Rule(org=sub_orgs['C-1'].org, sub_org=sub_orgs['C-1'], name='No Hats!'))
        session.add(Rule(org=sub_orgs['C-1'].org, name='No Shoes!'))
        session.add(Rule(org=orgs['A'], name='No ties!'))
        session.commit()


def main():
    engine = get_engine(os.environ)

    with engine.begin() as conn:
        Base.metadata.create_all(conn)

        populate(conn)

        query(conn)


if __name__ == '__main__':
    main()

A(1)
    Default Org Rules:
      #3 No ties!

    A-0(1)
        Sub Org Rules:
          #3 No ties!

B(2)
    Default Org Rules:

C(3)
    Default Org Rules:
      #2 No Shoes!

    C-0(2)
        Sub Org Rules:
          #2 No Shoes!

    C-1(3)
        Sub Org Rules:
          #1 No Hats!
          #2 No Shoes!