时间戳

首页 > 解决方案 > 如何在 SqlAlchemy 中进行没有 JOIN 的嵌套 SELECT？

python - 如何在 SqlAlchemy 中进行没有 JOIN 的嵌套 SELECT？

问题描述

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

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

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

现在我想使用这个“不透明”查询（本问题的黑盒）来构造其他查询，使用完全相同的标准从同一个表中，但在匹配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;

等等

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

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

标签： pythonpostgresqlsqlalchemy

解决方案

original_query只是一个SQLAlchemy 查询 API 对象，您可以对其应用其他过滤器和条件。查询 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)执行查询，但使用相同的过滤器。类似地，通过使用，我们只为该单列提供了一个新的查询对象，但过滤器仍然是该新查询的一部分。DISTINCT foo.barBETWEEN.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)列也将被选中。

推荐阅读