Sqlalchemy 对字典列表的查询

Question

我正在尝试运行一个根据字典列表过滤结果的查询。

让我们更清楚...

我有一个表格 Heading，每个 Heading 都有一个版本字段

class Heading(Base):
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    version = Column(Integer)

另一方面，我有一个如下所示的 dict 列表（Json 格式）：

"headings": [
    {
        "id": 1,
        "version": 1
    },
    {
        "id": 2,
        "version": 1
    }
]

我想对 Heading 表执行查询，排除数据库中的版本等于 (json) 列表中的版本的行。

我尝试执行如下操作：

res = Heading.query.filter(~Heading.id.in_([d['id'] for d in headings if d['version'] == Heading.version])).all()

数据库看起来像这样（Json 再次格式化，抱歉）：

[
{
    "id": 1,
    "name": "Heading1",
    "version": 2,
},
{
    "id": 2,
    "name": "Heading2",
    "version": 1,
}
]

但无论如何它都会返回 Heading2 行。基本上，我希望查询返回 Json 中未定义的所有行以及数据库中具有较新版本的行。

我不知道我的问题是否足够清楚。 也许这不是正确的方法，请告诉我

提前感谢您的帮助

编辑： 生成的 Raw SQL 是这样的：

SELECT heading.id, heading.name, heading.version
FROM heading
WHERE heading.id = heading.id

所以 sqlalchemy 查询有一个明显的问题，但我不知道是什么...

我认为它来自部分

if d['version'] == Heading.version

但我不知道如何在查询的其他地方重新表述这个条件

Answer 1

正如您所怀疑的，问题出在

[... for ... if d['version'] == Heading.version]

虽然d['version'] == Heading.version 产生了一个SQL 表达式对象，Python 然后使用它的真值进行过滤：

In [9]: bool(headings[0]['version'] == Heading.version)
Out[9]: False

所以你的列表理解结果是一个空列表。 Heading.id 永远不会出现在空列表中，所以

~Heading.id.in_([])

将简单地评估为真：

In [10]: print(~Heading.id.in_([]))
1 = 1

鉴于您的数据库支持composite IN queries，解决方案是与 (id, version) 元组进行比较：

In [9]: from sqlalchemy import tuple_

In [10]: session.query(Heading).\
    ...:     filter(~tuple_(Heading.id, Heading.version).in_(
    ...:         [(d['id'], d['version']) for d in headings])).\
    ...:     all()
Out[10]: [<__main__.Heading at 0x7f840cc8ae48>]

In [11]: _[0].name
Out[11]: 'Heading1'