如何在SQLAlchemy/SQLite中结合alias()使用json_each查询？

问题分析 #

你在SQLite命令行中能正常执行的查询，是从dataset表的samplesJSON数组里，筛选出instruction字段为intrus2的元素。但用SQLAlchemy实现时出现alias() got an unexpected keyword argument 'flat'错误，核心问题有两个：

• 为json_each结果设置别名的方式不符合SQLAlchemy语法
• Filter条件逻辑与原生查询不符，错误地直接判断jsonfield.value等于筛选值

正确实现方式 #

方式一：使用SQLAlchemy原生API（推荐） #

先确保Dataset模型的samples字段定义为JSON类型：

from sqlalchemy import Column, Integer, JSON, func
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Dataset(Base):
    __tablename__ = 'dataset'
    id = Column(Integer, primary_key=True)
    samples = Column(JSON)

再编写查询代码：

filter_val = 'intrus2'

# 为json_each的返回结果创建别名
jsonfield = func.json_each(Dataset.samples).alias('jsonfield')

# 构建查询：筛选出instruction为intrus2的JSON元素值
query_samples = dbSession.query(jsonfield.c.value)\
    .select_from(Dataset, jsonfield)\
    .filter(func.json_extract(jsonfield.c.value, '$.instruction') == filter_val)

# 执行查询获取结果
results = query_samples.all()

方式二：直接使用原生SQL文本（简单直接） #

如果觉得原生API写法繁琐，可以直接复用SQLite命令行的查询逻辑，通过text对象实现：

filter_val = 'intrus2'

query_samples = dbSession.query(text("jsonfield.value"))\
    .select_from(text("dataset, json_each(dataset.samples) as jsonfield"))\
    .filter(text("json_extract(jsonfield.value, '$.instruction') = :filter_val"))\
    .params(filter_val=filter_val)

results = query_samples.all()

错误原因说明 #

1. 别名创建错误：你之前用alias(func.json_each(Dataset.samples), 'jsonfield')的写法不符合SQLAlchemy规范，正确的方式是在函数调用后直接追加.alias('jsonfield')，或者用alias(func_obj, name='jsonfield')传入name参数指定别名。
2. Filter条件错误：原生查询是判断json_extract解析出的instruction字段等于intrus2，但你之前的代码直接判断jsonfield.value等于筛选值，逻辑完全不符，即使别名正确也无法得到预期结果。

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