在 sqlalchemy 中的有序子查询上加入 DISTINCT ON

Question

这是我的问题（一个极其简化的版本）。

我使用 Postgresql 作为后端并尝试构建 sqlalchemy 查询 来自另一个查询。

表设置

以下是示例中包含一些随机数据的表格。 您可以假设每个表都是在 sqlalchemy 中以声明方式声明的，使用 映射器的名称分别是 Item 和 ItemVersion。 在问题的最后，您可以找到我放置代码的链接 此问题中的所有内容，包括表定义。

一些项目。

item
+----+
| id |
+----+
|  1 |
|  2 |
|  3 |
+----+

包含每个项目版本的表格。每个都有至少一个。

item_version
+----+---------+---------+-----------+
| id | item_id | version | text      |
+----+---------+---------+-----------+
|  1 |       1 |       0 | item_1_v0 |
|  2 |       1 |       1 | item_1_v1 |
|  3 |       2 |       0 | item_2_v0 |
|  4 |       3 |       0 | item_3_v0 |
+----+---------+---------+-----------+

查询

现在，对于Item 上的给定 sqlalchemy 查询，我想要一个返回的函数 另一个查询，但这次是(Item, ItemVersion)，Items 所在的位置 与原始查询中的相同（并且顺序相同！），并且 ItemVersion 是每个Item 对应的最新版本。

这是一个 SQL 示例，非常简单：

首先对item 表进行随机查询

SELECT item.id as item_id
FROM item
WHERE item.id != 2
ORDER BY item.id DESC

对应

+---------+
| item_id |
+---------+
|       3 |
|       1 |
+---------+

然后从那个查询中，如果我想加入正确的versions，我可以做到

SELECT sq2.item_id AS item_id,
       sq2.item_version_id AS item_version_id,
       sq2.item_version_text AS item_version_text
FROM (
    SELECT DISTINCT ON (sq.item_id)
           sq.item_id AS item_id,
           iv.id AS item_version_id,
           iv.text AS item_version_text
    FROM (
        SELECT item.id AS item_id
        FROM item
        WHERE id != 2
        ORDER BY id DESC) AS sq
    JOIN item_version AS iv
      ON iv.item_id = sq.item_id
    ORDER BY sq.item_id, iv.version DESC) AS sq2
ORDER BY sq2.item_id DESC

Note that it has to be wrapped in a subquery a second time because the DISTINCT ON discards the ordering.

现在的挑战是编写一个在 sqlalchemy 中执行此操作的函数。 这是我目前所拥有的。

首先对项目进行初始 sqlalchemy 查询：

session.query(Item).filter(Item.id != 2).order_by(desc(Item.id))

然后我可以构建我的第二个查询，但没有原始顺序。在 换句话说，我不知道如何进行我所做的第二个子查询包装 SQL 来取回被DISTINCT ON 丢弃的排序。

def join_version(session, query):
    sq = aliased(Item, query.subquery('sq'))

    sq2 = session.query(sq, ItemVersion) \
        .distinct(sq.id) \
        .join(ItemVersion) \
        .order_by(sq.id, desc(ItemVersion.version))
    return sq2

我认为this SO question 可能是答案的一部分，但我不完全是 确定如何。

运行此问题中所有内容的代码（数据库创建、填充和 到目前为止我所拥有的失败的单元测试）can be found here。一般 如果你能修复join_version函数，它应该可以通过测试！

Answer 1

好的，所以我找到了一种方法。这有点骇人听闻，但仍然只查询数据库两次，所以我想我会活下来！基本上我首先在数据库中查询Items，然后我对ItemVersions 进行另一个查询，过滤item_id，然后是reordering with a trick I found here (this is also relevant)。

代码如下：

def join_version(session, query):                                  
    items = query.all()                                            
    item_ids = [i.id for i in items]                               
    items_v_sq = session.query(ItemVersion) \                      
        .distinct(ItemVersion.item_id) \                           
        .filter(ItemVersion.item_id.in_(item_ids)) \               
        .order_by(ItemVersion.item_id, desc(ItemVersion.version)) \
        .subquery('sq')                                            
    sq = aliased(ItemVersion, items_v_sq)                          
    items_v = session.query(sq) \                                  
        .order_by('idx(array{}, sq.item_id)'.format(item_ids))     
                                                                   
    return zip(items, items_v)