假设我有这些定义层次结构的表
组织
子组织
现在有另一个表引用两者:
规则
规则表的工作方式是,如果 sub_org_id 为空,则规则适用于整个组织。如果 sub_ord_id 不为空,则该规则将覆盖该特定 sub_org。
如何将其映射到 sqlalchemy 关系中?
谢谢
我只会使用常规查询来解决规则。 也许提供有关该问题的更多信息。
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!