sqlalchemy case when分组查询统计_综合

1.需求

根据过滤条件将设备按升级状态分组，统计总数和各个状态的数量。

2.原始数据

在这里插入图片描述

3.原生sql语句

select count(*) as "total",sum(case when upgrade_status='failed' then 1 else 0 end) as "failed_count",sum(case when upgrade_status='unstarted' then 1 else 0 end) as "unstarted_count",sum(case when upgrade_status='cancelled' then 1 else 0 end) as "cancelled_count",sum(case when upgrade_status='upgrading' then 1 else 0 end) as "upgrading_count",sum(case when upgrade_status='succeed' then 1 else 0 end) as "succeed_count"
from edge_ota_record 
where task_id='73860eaffb014dba80db2f67518d5994' 
GROUP BY upgrade_status;

结果：
在这里插入图片描述

3.使用sqlalchemy

from sqlalchemy import func, casestatistics_dict = {
    }
qs = self.session.query(func.count().label("total"),func.sum(case(whens=[(self.model.upgrade_status == 'unstarted', 1)],else_=0)).label("unstarted_count"),func.sum(case(whens=[(self.model.upgrade_status == 'succeed', 1)],else_=0)).label("succeed_count"),func.sum(case(whens=[(self.model.upgrade_status == 'failed', 1)],else_=0)).label("failed_count"),func.sum(case(whens=[(self.model.upgrade_status == 'cancelled', 1)],else_=0)).label("cancelled_count"),func.sum(case(whens=[(self.model.upgrade_status == 'upgrading', 1)],else_=0)).label("upgrading_count"),).filter(self.model.task_id == task_id).group_by(self.model.upgrade_status)results = [dict(zip(result.keys(), result)) for result in qs.all()]counter = collections.Counter()
for result_dict in results:counter.update(result_dict)
statistics_dict.update(dict(counter))print(statistics_dict )

结果：

 # {"total": 18, "unstarted_count": 18, "upgrading_count": 0, "cancelled_count": 0, "failed_count": 0, "succeed_count": 0}