python - SQLAlchemy:查找数组列之间的差异

Question

我有一个名为 UnitOfWork 的表，它有 3 列 cases_identified、cases_completed 和 cases_double_check，所有这些是整数的 Postgresql 数组。

是否可以编写一个查询(或混合属性)来查找已识别但不在已完成或双重检查列中的案例？

这是我想出来的，但是 SQL 表达式不起作用:

@hybrid_property
def todo(self):
    return [x for x in self.cases_identified
            if x not in self.cases_completed and
            x not in self.cases_double_check]

@todo.expression
def todo(cls):
    return [x for x in cls.cases_identified
            if x not in  cls.cases_completed and
            x not in cls.cases_double_check]

我在测试查询中得到的错误是:

test = Session.query(UnitOfWork.todo).first()
NotImplementedError: Operator 'contains' is not supported on this expression

Answer 1

对于这个答案，我假设 cls.cases_identified、cls.cases_completed 和 cls.cases_double_check 属于 postgresql .ARRAY(Integer) 在 Python 端输入。

您的 @todo.expression 应该只返回:一个 SQL 表达式。目前它正在尝试返回一个 python 列表。自 postgresql.ARRAY 以来引发异常不支持 in 运算符，尽管它有一个方法 contains ，它映射到 Postgresql 中的 @> 运算符并测试“元素是否是参数数组表达式元素的超集”。另一方面，这不是您想要的。你很幸运，你在那里有 if x not in ...，很简单

[x for x in cls.cases_identified]

似乎导致无限循环而不是异常。

获取 Postgresql 中数组之间的差异已涵盖 here广泛，但这里是你如何使用 SQLAlchemy 应用它，首先使用 an array constructor :

from sqlalchemy import func

...

class UnitOfWork(...):

    @todo.expression
    def todo(cls):
        # Force correlation of `cls` from outer FROM objects. Use
        # the `select()` shorthand method of `FunctionElement`.
        identified = func.unnest(cls.cases_identified).select().correlate(cls)
        completed = func.unnest(cls.cases_completed).select().correlate(cls)
        double_check = func.unnest(cls.cases_double_check).select().correlate(cls)
        # Create the subquery statement
        stmt = identified.except_(completed.union(double_check))
        # Uses the `func` generic for ARRAY constructor from
        # subquery syntax. The subquery statement has to be
        # converted to a "scalar subquery" in the eyes of SQLA
        # (really just a subquery returning 1 col) in order
        # to remove it from the FROM clause hierarchy.
        return func.array(stmt.as_scalar())

这有一个不提供任何 FROM 对象的缺点(因为它与封闭查询中的所有内容相关联)，因此您必须像这样发出原始查询:

test = Session.query(UnitOfWork.todo).select_from(UnitOfWork).first()

您还可以使用 Postgresql intarray module为无空整数数组提供特殊函数和运算符:

class UnitOfWork(...):

    @todo.expression
    def todo(cls):
        return (cls.cases_identified - cls.cases_completed - 
                cls.cases_double_check)

请注意，您必须先在 Postgresql 中安装扩展:

CREATE EXTENSION intarray;