Using 'case' in a select query sqlalchemy throwing exception

Question

I'm trying to construct this sql query into 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

My attempt:

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
)

When I try this, I get the exception:

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

From some digging, it is unhappy about this line:

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

Answer 1

You are not using the case(...) properly - you need to put your condition in a tuple of (condition, value) instead so that:

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

becomes:

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

The same applies for the other case statement.

---

Furthermore, your filter condition audit_table.c.audited_at is not None will not result in the query you are trying to build. This expression will evaluate to boolean value of True and therefore will generate a truthy WHERE clause such as WHERE 1=1 (or nothing at all).
Instead you should use audit_table.c.audited_at != None. The PEP-8 checks will complain, but this is how it should be used with sqlalchemy.