python - SQLAlchemy ORM 从子查询中选择多个实体

Question

我需要session.query(Entity1, Entity2)仅从子查询而不是直接从表中查询多个实体，例如 。文档有一些关于从子查询中选择一个实体的内容，但我找不到如何选择多个实体，无论是在文档中还是通过实验。

我的用例是我需要通过窗口函数过滤映射类底层的表，这在 PostgreSQL 中只能在子查询或 CTE 中完成。

编辑：子查询跨越两个表的 JOIN 所以我不能只做aliased(Entity1, subquery).

Answer 1

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    bs = relationship("B")

class B(Base):
    __tablename__ = "b"

    id = Column(Integer, primary_key=True)

    a_id = Column(Integer, ForeignKey('a.id'))

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
s = Session(e)
s.add_all([A(bs=[B(), B()]), A(bs=[B()])])
s.commit()

# with_labels() here is to disambiguate A.id and B.id.
# without it, you'd see a warning
# "Column 'id' on table being replaced by another column with the same key."
subq = s.query(A, B).join(A.bs).with_labels().subquery()


# method 1 - select_from()
print s.query(A, B).select_from(subq).all()

# method 2 - alias them both.  "subq" renders
# once because FROM objects render based on object
# identity.
a_alias = aliased(A, subq)
b_alias = aliased(B, subq)
print s.query(a_alias, b_alias).all()

Answer 2

我试图做类似原始问题的事情：使用外连接将过滤表与另一个过滤表连接起来。我一直在苦苦挣扎，因为如何：

• 创建一个从两个表中返回实体的 SQLAlchemy 查询。@zzzeek 的回答向我展示了如何做到这一点：get_session().query(A, B)。
• 在此类查询中将查询用作表。@zzzeek 的回答也向我展示了如何做到这一点：filtered_a = aliased(A).filter(...).subquery()。
• 在两个实体之间使用 OUTER 连接。使用select_from()afterouterjoin()会破坏表之间的连接条件，从而导致交叉连接。从@zzzeek 回答我猜如果a是别名（），那么您可以包含a在查询（）和.outerjoin（a）中，它不会再次加入，这似乎有效。

遵循@zzzeek 建议的任何一种方法都直接导致交叉连接（组合爆炸），因为我的一个模型使用继承，并且 SQLAlchemy 在内部 SELECT 之外添加了父表，没有任何条件！我认为这是 SQLAlchemy 中的一个错误。我最终采用的方法是：

filtered_a = aliased(A, A.query().filter(...)).subquery("filtered_a")
filtered_b = aliased(B, B.query().filter(...)).subquery("filtered_b")
query = get_session().query(filtered_a, filtered_b)
query = query.outerjoin(filtered_b, filtered_a.relation_to_b)
query = query.order_by(filtered_a.some_column)

for a, b in query:
    ...