如何关闭SQLAlchemy中SQLite JSON查询的JSON_QUOTE自动引用？

问题背景 #

使用SQLAlchemy操作JSON字段时，MySQL的实现简洁直观，但SQLite会自动给JSON_EXTRACT的结果套上JSON_QUOTE()，导致必须手动用func.json_quote()才能匹配值，生成的SQL冗余且不直观。

通用模型代码 #

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, JSON

Base = declarative_base()

class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer(), primary_key=True)
    message = Column(JSON)

MySQL的简洁实现 #

直接通过字段路径匹配字符串值即可，生成的SQL清晰：

engine = create_engine('mysql+pymysql://root:123456@localhost:3306/test', pool_recycle=3600)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

person = Person(message={'name': 'Bob'})
session.add(person)
session.commit()

# 直接匹配字符串，无需额外处理
query = session.query(Person.id).filter(Person.message['name'] == 'Bob')
print(query[0].id)  # 输出1
# 生成的SQL：
# SELECT person.id AS person_id 
# FROM person 
# WHERE JSON_EXTRACT(person.message, %(message_1)s) = %(param_1)s

SQLite的冗余问题 #

默认情况下，SQLite会自动给JSON提取结果添加JSON_QUOTE()，必须手动给匹配值套func.json_quote()才能生效：

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

person = Person(message={'name': 'Bob'})
session.add(person)
session.commit()

# 必须手动套json_quote才能匹配
query = session.query(Person.id).filter(Person.message['name'] == func.json_quote('Bob'))
print(query[0].id)  # 输出1
# 生成的SQL：
# SELECT person.id AS person_id 
# FROM person 
# WHERE JSON_QUOTE(JSON_EXTRACT(person.message, ?)) = json_quote(?)

解决方法：关闭SQLite的自动JSON_QUOTE #

以下两种方法可以实现和MySQL一致的简洁查询逻辑：

方法1：自定义无自动引用的JSON类型 #

继承原生JSON类型，重写SQLite方言的比较逻辑：

from sqlalchemy import JSON as _JSON

class UnquotedJSON(_JSON):
    def bind_expression(self, bindvalue):
        # 直接返回绑定值，不套JSON_QUOTE
        return bindvalue

    class comparator_factory(_JSON.comparator_factory):
        def __eq__(self, other):
            # 重写等于比较，不给提取结果加JSON_QUOTE
            if isinstance(other, str):
                return self.expr == other
            return super().__eq__(other)

# 修改模型字段为自定义类型
class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer(), primary_key=True)
    message = Column(UnquotedJSON)

之后在SQLite中即可像MySQL一样直接查询：

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

person = Person(message={'name': 'Bob'})
session.add(person)
session.commit()

# 无需手动套json_quote
query = session.query(Person.id).filter(Person.message['name'] == 'Bob')
print(query[0].id)  # 输出1
# 生成的SQL：
# SELECT person.id AS person_id 
# FROM person 
# WHERE JSON_EXTRACT(person.message, ?) = ?

方法2：全局修改SQLite JSON方言的比较逻辑 #

如果不想自定义类型，可以直接修改SQLAlchemy内置的SQLite JSON方言行为，全局生效：

from sqlalchemy.dialects.sqlite import JSON as SQLiteJSON

# 保存原生的等于比较方法
original_eq = SQLiteJSON.Comparator.__eq__

def custom_eq(self, other):
    # 字符串值直接比较，不套JSON_QUOTE
    if isinstance(other, str):
        return self.expr == other
    return original_eq(self, other)

# 替换原生方法
SQLiteJSON.Comparator.__eq__ = custom_eq

修改后使用原生JSON字段类型即可实现简洁查询，效果和方法1一致。

注意事项 #

• 以上修改仅对SQLite生效，不会影响MySQL等其他数据库的JSON查询逻辑
• 如果JSON字段值为数字、布尔值等非字符串类型，需根据实际场景调整比较逻辑，避免类型不匹配

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