python - 如何在 SqlAlchemy 中进行不带 JOIN 的嵌套 SELECT？

Question

我有一个 Postgres 查询(通过 SQLAlchemy)，它使用复杂的条件选择匹配的行:

original_query = session.query(SomeTable).filter(*complex_filters)

我不知道查询是如何构造的，我只能访问生成的查询实例。

现在我想使用这个“不透明”查询(出于这个问题的目的是黑盒)来构造其他查询，从同一个表使用完全相同的标准，但在匹配的 original_query 行之上有额外的逻辑.例如，顶部为 SELECT DISTINCT(column):

another_query = session.query(SomeTable.column).distinct().?select_from_query?(original_query)

或

SELECT SUM(tab_value) FROM (
    SELECT tab.key AS tab_key, tab.value AS tab_value -- inner query, fixed
    FROM tab
    WHERE tab.product_id IN (1, 2)  -- simplified; the inner query is quite complex
) AS tbl
WHERE tab_key = 'length';

或

SELECT tab_key, COUNT(*) FROM (
    SELECT tab.key AS tab_key, tab.value AS tab_value
    FROM tab
    WHERE tab.product_id IN (1, 2)
) AS tbl
GROUP BY tab_key;

等等

如何在 SQLAlchemy 中干净地实现 ?select_from_query? 部分？基本上，如何在 SqlAlchemy 中执行 SELECT dynamic FROM (SELECT fixed)？

---

动机:内部 Query 对象来自代码的不同部分。我无法控制它的构造方式，并且希望避免为我必须在其上运行的每个 SELECT 重复其逻辑。我想重新使用该查询，但在顶部添加额外的逻辑(按照上面的示例)。

Answer 1

original_query只是一个SQLAlchemy query API object ，您可以对此应用其他过滤器和条件。查询 API 是生成的；每个Query()实例操作返回一个新的(不可变的)实例并且您的起点(original_query)不受影响。

这包括使用 Query.distinct() 添加 DISTINCT()条款， Query.with_entities() 更改哪些列是查询的一部分，以及 Query.values() 执行查询但仅返回特定的单列值。

使用 .distinct(<column>).with_entities(<column>)创建一个新的查询对象(可以进一步重复使用):

another_query = original_query.distinct(SomeTable.column).with_entities(SomeTable.column)

或者只使用 .distinct(<column>).values(<column>)得到 (column_value,) 的迭代器元组结果就在那里然后:

distinct_values = original_query.distinct(SomeTable.column).values(SomeTable.column)

请注意 .values()立即执行查询，如 .all()会，而.with_entities()还给你一个新的 Query只有单列的对象(然后将执行 .all() 或迭代或切片并返回结果)。

演示，使用人为的 Foo模型(针对 sqlite 执行以使其更容易快速演示):

>>> from sqlalchemy import *
>>> from sqlalchemy.ext.declarative import declarative_base
>>> from sqlalchemy.orm import sessionmaker
>>> Base = declarative_base()
>>> class Foo(Base):
...     __tablename__ = "foo"
...     id = Column(Integer, primary_key=True)
...     bar = Column(String)
...     spam = Column(String)
...
>>> engine = create_engine('sqlite:///:memory:', echo=True)
>>> session = sessionmaker(bind=engine)()
>>> Base.metadata.create_all(engine)
2019-06-10 13:10:43,910 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("foo")
2019-06-10 13:10:43,910 INFO sqlalchemy.engine.base.Engine ()
2019-06-10 13:10:43,911 INFO sqlalchemy.engine.base.Engine
CREATE TABLE foo (
    id INTEGER NOT NULL,
    bar VARCHAR,
    spam VARCHAR,
    PRIMARY KEY (id)
)


2019-06-10 13:10:43,911 INFO sqlalchemy.engine.base.Engine ()
2019-06-10 13:10:43,913 INFO sqlalchemy.engine.base.Engine COMMIT
>>> original_query = session.query(Foo).filter(Foo.id.between(17, 42))
>>> print(original_query)  # show what SQL would be executed for this query
SELECT foo.id AS foo_id, foo.bar AS foo_bar, foo.spam AS foo_spam
FROM foo
WHERE foo.id BETWEEN ? AND ?
>>> another_query = original_query.distinct(Foo.bar).with_entities(Foo.bar)
>>> print(another_query)  # print the SQL again, don't execute
SELECT DISTINCT foo.bar AS foo_bar
FROM foo
WHERE foo.id BETWEEN ? AND ?
>>> distinct_values = original_query.distinct(Foo.bar).values(Foo.bar)  # executes!
2019-06-10 13:10:48,470 INFO sqlalchemy.engine.base.Engine SELECT DISTINCT foo.bar AS foo_bar
FROM foo
WHERE foo.id BETWEEN ? AND ?
2019-06-10 13:10:48,470 INFO sqlalchemy.engine.base.Engine (17, 42)

在上面的演示中，原始查询将选择某些 Foo带有 BETWEEN 的实例过滤器，但添加 .distinct(Foo.bar).values(Foo.bar)然后对 just DISTINCT foo.bar 执行查询列，但具有相同的 BETWEEN过滤到位。同样，通过使用 .with_entities() ，我们只为该单列提供了一个新的查询对象，但过滤器仍然是该新查询的一部分。

您添加的示例的工作方式相同；您实际上不需要在那里进行子选择，因为相同的查询可以表示为:

SELECT sum(tab.value)
FROM tab
WHERE tab.product_id IN (1, 2) AND tab_key = 'length';

这可以简单地通过添加额外的过滤器然后使用 .with_entities() 来实现。用您的 SUM() 替换所选的列:

summed_query = (
    original_query
    .filter(Tab.key == 'length')  # add a filter
    .with_entities(func.sum(Tab.value)

或者，根据上述Foo演示:

>>> print(original_query.filter(Foo.spam == 42).with_entities(func.sum(Foo.bar)))
SELECT sum(foo.bar) AS sum_1
FROM foo
WHERE foo.id BETWEEN ? AND ? AND foo.spam = ?

有子查询的用例(例如限​​制连接中特定表的结果)，但这不是其中之一。

如果你确实需要一个子查询，那么查询 API 有 Query.from_self() (对于更简单的情况)和 Query.subselect() .

例如，如果您只需要从原始查询中选择聚合行并通过 HAVING 过滤聚合值, 然后将结果与另一个表连接起来，以获得每个组的最高行 ID 和一些进一步的过滤，然后你需要一个子查询:

summed_col = func.sum(SomeTable.some_column)
max_id = func.max(SomeTable.primary_key)
summed_results_by_eggs = (
    original_query
    .with_entities(max_id, summed_col)      # only select highest id and the sum
    .group_by(SomeTable.other_column)       # per group
    .having(summed_col > 10)                # where the sum is high enough
    .from_self(summed_col)                  # give us the summed value as a subselect
    .join(                                  # join these rows with another table
        OtherTable,
        OtherTable.foreign_key == max_id    # using the highest id
    )
    .filter(OtherTable.some_column < 1000)  # and filter some more
)

上面只会选择求和的SomeTable.some_column该值大于 10 且最高 SomeTable.id 的值每个组中的值。此查询必须使用子查询，因为您要限制符合条件的 SomeTable加入另一个表之前的行。

为了演示这个，我添加了第二个表 Eggs :

>>> from sqlalchemy.orm import relationship
>>> class Eggs(Base):
...     __tablename__ = "eggs"
...     id = Column(Integer, primary_key=True)
...     foo_id = Column(Integer, ForeignKey(Foo.id))
...     foo = relationship(Foo, backref="eggs")
...
>>> summed_col = func.sum(Foo.bar)
>>> max_id = func.max(Foo.id)
>>> print(
...     original_query
...     .with_entities(max_id, summed_col)
...     .group_by(Foo.spam)
...     .having(summed_col > 10)
...     .from_self(summed_col)
...     .join(Eggs, Eggs.foo_id==max_id)
...     .filter(Eggs.id < 1000)
... )
SELECT anon_1.sum_2 AS sum_1
FROM (SELECT max(foo.id) AS max_1, sum(foo.bar) AS sum_2
FROM foo
WHERE foo.id BETWEEN ? AND ? GROUP BY foo.spam
HAVING sum(foo.bar) > ?) AS anon_1 JOIN eggs ON eggs.foo_id = anon_1.max_1
WHERE eggs.id < ?

Query.from_self()方法采用新实体在外部查询中使用，如果您省略这些实体，则所有列都将被拉出。在上面，我提取了汇总的列值；没有那个论点MAX(Foo.id)列也将被选中。