SqlAlchemy：使用双存在（或）和附加布尔检查进行子选择

Question

在SqlAlchemy中，我需要实现以下子查询，它在PostgreSQL中运行良好。它是一个OR条件，由2个EXISTS和一个附加AND块组成。整个列将产生一个真/假布尔值

SELECT
...
...
,

(SELECT 
    (
        EXISTS (SELECT id from participating_ic_t pi1 where pi1.id = agreement_t_1.participating_ic_id 
                AND pi1.ic_nihsac = substr(ned_person_t_2.nihsac, 1, 3))
     OR EXISTS (SELECT id from external_people_t ep1 where ep1.participating_ic_id = agreement_t_1.participating_ic_id 
                AND ep1.uniqueidentifier = ned_person_t_2.uniqueidentifier)
    )
    AND ned_person_t_2.current_flag = 'Y' and ned_person_t_2.inactive_date is null and ned_person_t_2.organizationalstat = 'EMPLOYEE'
) as ACTIVE_APPROVER1,

首先，如果省略附加AND块，则以下OR-EXISTS本身可以正常工作：

subq1 = db_session.query(ParticipatingIcT.id).filter((ParticipatingIcT.id == agreement.participating_ic_id),
                                                     (ParticipatingIcT.ic_nihsac == func.substr(approver.nihsac, 1, 3)))
                                             .subquery()
subq2 = db_session.query(ExternalPeopleT.id).filter((ExternalPeopleT.participating_ic_id == agreement.participating_ic_id),
                                                    (ExternalPeopleT.uniqueidentifier == approver.uniqueidentifier))
                                            .subquery()                                     
subqMain =  db_session.query(or_(exists(subq1), exists(subq2))
                      .subquery()

# ...
# Now we will select from subqMain.                   
agreements = (db_session.query(
                   ..,
                   subqMain
               

但是当我介绍最后一个和块时，问题就开始了。从概念上讲，最终结果应如下所示：

subqMain = db_session.query(and_(
                                or_(exists(subq1), exists(subq2)), 
                                approver.current_flag == 'Y',
                                approver.inactive_date == None, 
                                approver.organizationalstat == 'EMPLOYEE'))
                      .subquery()

但是它实际上在子查询的右边发出" .. FROM APPROVER_T"，而它应该在最末端链接到主查询的FROM APPROVER_T。我需要避免添加.. FROM [table]，这会在我指定and_(..)后立即发生。我不明白它为什么这么做。所有子查询都专门标记为subquery()。别名approver在最顶端定义为approver = aliased(NedPersonT)

Answer 1

exists1 = db_session.query(
    ParticipatingIcT.id
).filter(
    (ParticipatingIcT.id == agreement.participating_ic_id),
    (ParticipatingIcT.ic_nihsac == func.substr(approver.nihsac, 1, 3))
).exists() # use exists here instead of subquery
exists2 = db_session.query(
    ExternalPeopleT.id
).filter(
    (ExternalPeopleT.participating_ic_id == agreement.participating_ic_id),                                                  
    (ExternalPeopleT.uniqueidentifier == approver.uniqueidentifier)
).exists()  # use exists again instead of subquery                                     

subqMain =  db_session.query(and_(
    or_(exists1, exists2),
    approver.current_flag == 'Y',
    approver.inactive_date == None, 
    approver.organizationalstat == 'EMPLOYEE')).subquery()

# ...
# Now we will select from subqMain.                   
agreements = (db_session.query(
                   OtherModel,
                   subqMain))

sqlalchemy.orm.Query.exists

半生不熟的例子

我在这里也使用别名，因为我不太了解将子查询包装在括号中的操作方法。如果没有它，它似乎可以工作，但是原始sql很难读取。这是我制作的一个玩具示例，用于尝试查看生成的SQL是否无效

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    hobbies = relationship('Hobby', backref='user')

class Hobby(Base):
    __tablename__ = 'hobbies'
    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable = False)
    name = Column(String, nullable=False)


Base.metadata.create_all(engine, checkfirst=True)
with Session(engine) as session:
    user1 = User(name='user1')
    session.add(user1)
    session.add_all([
        Hobby(name='biking', user=user1),
        Hobby(name='running', user=user1),
        Hobby(name='eating', user=user1),
    ])
    user2 = User(name='user2')
    session.add(user2)
    session.add(Hobby(name='biking', user=user2))
    session.commit()


    nested_user = aliased(User)
    subq1 = session.query(Hobby.id).filter(nested_user.id ==Hobby.user_id, Hobby.name.like('bik%'))
    subq2 = session.query(Hobby.id).filter(nested_user.id ==Hobby.user_id, Hobby.name.like('eat%'))
    subqmain = session.query(nested_user).filter(or_(subq1.exists(), subq2.exists()), nested_user.id > 0).subquery()
    q = session.query(User).select_from(User).join(subqmain, User.id == subqmain.c.id)
    print (q)
    print ([user.name for user in q.all()])

SELECT users.id AS users_id, users.name AS users_name 
FROM users JOIN (SELECT users_1.id AS id, users_1.name AS name 
FROM users AS users_1 
WHERE ((EXISTS (SELECT 1 
FROM hobbies 
WHERE users_1.id = hobbies.user_id AND hobbies.name LIKE %(name_1)s)) OR (EXISTS (SELECT 1 
FROM hobbies 
WHERE users_1.id = hobbies.user_id AND hobbies.name LIKE %(name_2)s))) AND users_1.id > %(id_1)s) AS anon_1 ON users.id = anon_1.id
['user1', 'user2']

Answer 2

该子选择不起作用，因此我将其改写为一个案例语句。现在它开始工作了。此外，我还必须在主查询中加入几个额外的表，以方便处理这个新情况

casePrimaryApprover = case(
    [
        (and_(
            (agreement.approving_official_id.is_not(None)),
            (approver.current_flag == 'Y'),
            (approver.inactive_date.is_(None)), 
            (approver.organizationalstat == 'EMPLOYEE'),
            or_(
                (participatingIc.ic_nihsac == func.substr(approver.nihsac, 1, 3)), 
                (externalPeoplePrimaryApprover.id.is_not(None))
            )
        ), 'Y')
    ],
    else_ = 'N'
)


# Main Query
agreements = (db_session.query(
    agreement.id,
    ...
    casePrimaryApprover 
    
    ...
    .join(participatingIc, agreement.participating_ic_id == participatingIc.id)
    .outerjoin(externalPeoplePrimaryApprover, approver.uniqueidentifier == externalPeoplePrimaryApprover.uniqueidentifier)