SQLAlchemy查询PostgreSQL含数组的JSON列并统计匹配项数量

你的查询返回结果不符合预期的核心原因是：safety_data字段存储的是JSON数组，而非单个JSON对象，不能直接按键索引查询，也不能仅查询固定下标的元素。

场景1：统计符合条件的Traydetails记录总数 #

即只要某条记录的safety_data数组中至少有1个Hazard Rating为Class B，就算1条记录：

from sqlalchemy import func, Unicode

hazard_record_count = Traydetails.query\
    .join(func.json_array_elements(Traydetails.safety_data).alias('safety_item'))\
    .filter(
        func.json_extract_path_text(func.column('safety_item'), 'Hazard Rating').cast(Unicode) == 'Class B'
    )\
    .distinct(Traydetails.id)\
    .count()

场景2：统计所有符合条件的JSON条目总数 #

即某条记录的safety_data数组中有N个Hazard Rating为Class B，就算N条：

from sqlalchemy import func, Unicode

hazard_item_count = Traydetails.query\
    .join(func.json_array_elements(Traydetails.safety_data).alias('safety_item'))\
    .filter(
        func.json_extract_path_text(func.column('safety_item'), 'Hazard Rating').cast(Unicode) == 'Class B'
    )\
    .count()

拓展：一次性统计所有风险等级的数量 #

如果需要同时统计Class A/Class B/Class N等所有等级的总数，可以直接用分组查询：

from sqlalchemy import func, Unicode

rating_stats = db.session.query(
    func.json_extract_path_text(func.column('safety_item'), 'Hazard Rating').cast(Unicode).label('rating'),
    func.count().label('count')
)\
.select_from(Traydetails)\
.join(func.json_array_elements(Traydetails.safety_data).alias('safety_item'))\
.group_by('rating')\
.all()

# 遍历即可拿到所有等级的统计结果
for rating, count in rating_stats:
    print(f"风险等级{rating}的总数量：{count}")

优化建议 #

如果后续查询频率较高，建议将模型中的JSON类型改为JSONB类型，PostgreSQL对JSONB的查询性能远高于普通JSON类型，仅需将代码中json_array_elements替换为jsonb_array_elements即可，其余逻辑完全兼容。

内容的提问来源于stack exchange，提问作者Manoj Selvam