如何用SQLAlchemy筛选PostgreSQL中含defect_type_id=1的JSON字段数据

针对你的SQLAlchemy模型和PostgreSQL数据库，有两种常用方式实现需求：

方法一：展开JSON数组+子查询判断 #

利用PostgreSQL的json_array_elements函数展开JSON数组，结合exists子查询检查是否存在符合条件的元素：

from sqlalchemy import exists, func
from sqlalchemy.types import Integer

# 构造子查询条件
has_target_defect = exists().where(
    # 展开defects_list数组，检查其中元素的defect_type_id是否为1
    func.json_array_elements(Main.defects_list).table_valued().columns[0]['defect_type_id'].astext.cast(Integer) == 1
)

# 执行查询
target_mains = session.query(Main).filter(has_target_defect).all()

如果JSON中的defect_type_id本身就是数字类型（如你的示例数据），可以省略astext.cast(Integer)，直接比较：

has_target_defect = exists().where(
    func.json_array_elements(Main.defects_list).table_valued().columns[0]['defect_type_id'] == 1
)

方法二：使用JSON路径查询（简洁高效） #

PostgreSQL支持JSON路径语法，通过jsonb_path_exists函数直接检查数组中是否存在满足条件的元素。如果你的字段是JSON类型，可以先转为JSONB（性能更优）：

from sqlalchemy import func
from sqlalchemy.dialects.postgresql import JSONB

# 构造路径查询条件
condition = func.jsonb_path_exists(
    Main.defects_list.cast(JSONB),
    '$.[] ? (@.defect_type_id == 1)'
)

# 执行查询
target_mains = session.query(Main).filter(condition).all()

如果你的defects_list字段本身定义为JSONB类型（推荐用于频繁JSON操作的场景），直接传入字段即可：

condition = func.jsonb_path_exists(
    Main.defects_list,
    '$.[] ? (@.defect_type_id == 1)'
)

说明 #

• 方法一适合需要对数组元素做额外处理（如过滤、聚合）的场景；
• 方法二语法更简洁，PostgreSQL对JSON路径查询有专门优化，大数据量下性能表现更好。

内容的提问来源于stack exchange，提问作者Sviatoslav Kalina