如何将Jsonb类型列的SQL查询添加到运算符列表中？

需求说明 #

需将针对JSONB类型列的查询逻辑添加到现有运算符字典，原SQL用于筛选rulesets表中满足以下条件的记录：

• rules（JSONB数组类型列）包含指定UUID列表中的任意值
• client_id='10001'
• is_deleted=false

原SQL查询 #

SELECT * FROM rulesets WHERE EXISTS (
    SELECT 1 FROM jsonb_array_elements_text(rulesets.rules) AS elem
    WHERE elem.value IN ('2c084b4e-9658-4f50-afe4-a0a820d1d51d', 
        '23de26ef-c3aa-4f31-81dd-1b63e1e2f699',
        '7a1af945-c5bd-4709-b5f3-24180398da7d', 
        '665ecd7e-900a-4b30-b6da-19ff7a0068b3')
) AND client_id='10001' AND is_deleted=false;

现有运算符列表 #

OPERATORS = {
    'is_null': lambda f, a: f.is_(None) if a else f.isnot(None),
    'eq': lambda f, a: f == a,
    'ne': lambda f, a: f != a,
    'gt': lambda f, a: f > a,
    'lt': lambda f, a: f < a,
    'gte': lambda f, a: f >= a,
    'lte': lambda f, a: f <= a,
    'like': lambda f, a: f.like(a),
    'ilike': lambda f, a: f.ilike(a),
    'not_ilike': lambda f, a: ~f.ilike(a),
    'in': lambda f, a: f.in_(a),
    'list_in': lambda f, a: f.contains(a),
    'or': lambda f, a: f.or_(a),
    'not_in': lambda f, a: ~f.in_(a),
    'nin': lambda f, a: ~f.in_(a),
    'any': lambda f, a: f.any(a),
    'not_any': lambda f, a: func.not_(f.any(a)),
}

错误尝试的自定义运算符 #

'custom_query': lambda f, a: f.exists().where(and_(exists().where(jsonb_array_elements_text(f.rules).in_(a)), f.client_id == '10001', f.is_deleted == False)),

正确解决方案 #

步骤1：确保导入必要依赖 #

from sqlalchemy import exists, and_, func
from sqlalchemy.dialects.postgresql import jsonb_array_elements_text

步骤2：添加正确的运算符逻辑 #

如果client_id和is_deleted为固定值：

OPERATORS['rules_contains_any'] = lambda f, a: and_(
    exists(
        select(jsonb_array_elements_text(f.c.rules))
        .where(jsonb_array_elements_text(f.c.rules).in_(a))
    ),
    f.c.client_id == '10001',
    f.c.is_deleted == False
)

如果需要动态传入client_id和is_deleted，可调整为接收字典参数：

OPERATORS['rules_contains_any'] = lambda f, params: and_(
    exists(
        select(jsonb_array_elements_text(f.c.rules))
        .where(jsonb_array_elements_text(f.c.rules).in_(params['uuids']))
    ),
    f.c.client_id == params['client_id'],
    f.c.is_deleted == params['is_deleted']
)

逻辑说明 #

• 用exists构造子查询，模拟原SQL中jsonb_array_elements_text展开数组并匹配值的逻辑
• 通过and_组合所有筛选条件，与原SQL逻辑完全对齐
• 若上下文里f直接是列对象而非表对象，需替换为对应的表引用（比如rulesets.c.rules）

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