python - 在选择查询中使用 'case' sqlalchemy 抛出异常

Question

我正在尝试将此 sql 查询构建到 sqlalchemy 中:

select distinct
    processed_by_user_uuid,
    sum(case when audited_result_uuid = PASSED_RESULT_UUID then 1 else 0 end) as "passed",
    sum(case when audited_result_uuid <> PASSED_RESULT_UUID then 1 else 0 end) as "failed",
    count(distinct processed_using) as "total audits"
from audit
where
    audited_at is not null
    and processed_at > [start_time]
    and processed_at < [end_time]
group by 1

我的尝试:

query = Session.query(audit_table).with_entities(
    audit_table.c.processed_by_user_uuid,
    func.sum(
        case([audit_table.c.audited_result_uuid == PASSED_RESULT_UUID, 1], else_=0)
    ).label('passed'),
    func.sum(
        case([audit_table.c.audited_result_uuid != PASSED_RESULT_UUID, 1], else_=0)
    ).label('failed'),
    func.count(distinct(audit_table.c.processed_using)).label('total audits')
).filter(
    audit_table.c.audited_at is not None,
    audit_table.c.processed_at > start_time,
    audit_table.c.processed_at < end_time
).group_by(
    1
)

当我尝试这个时，我得到了异常:

NotImplementedError: Operator 'getitem' is not supported on this expression

从一些挖掘来看，这条线令人不快:

func.sum(
    case([audit_table.c.audited_result_uuid == PASSED_RESULT_UUID, 1], else_=0)
).label('passed')

Answer 1

您没有使用 case(...)正确地 - 你需要把你的条件放在 (condition, value) 的元组中，这样:

case([audit_table.c.audited_result_uuid == PASSED_RESULT_UUID, 1], else_=0)

变成:

case([(audit_table.c.audited_result_uuid == PASSED_RESULT_UUID, 1)], else_=0)

这同样适用于其他 case 语句。

---

此外，您的过滤条件 audit_table.c.audited_at is not None 不会导致您尝试构建的查询。此表达式将评估为 True 的 boolean 值，因此将生成 truthy WHERE 子句，例如 WHERE 1=1(或什么都没有)。
相反，您应该使用 audit_table.c.audited_at != None。 PEP-8 检查会报错，但这是它应该如何与 sqlalchemy 一起使用。