如何在SqlAlchemy中查询TEXT类型列中的有效JSON数据？

因为你的data列是TEXT类型而非原生JSON列，直接用data["lastname"]的语法会无法识别，需要先将TEXT内容转换为JSON类型再进行查询。以下分场景给出实现方案：

分数据库针对性实现 #

PostgreSQL #

利用PostgreSQL的jsonb/json函数转换TEXT，结合astext获取字符串值：

from sqlalchemy import func

records = db_session.query(Resource).filter(
    func.jsonb(Resource.data)["lastname"].astext == "Doe"
).all()

如果想贴近你期望的写法，也可以给模型加一个JSON类型的属性：

from sqlalchemy import type_coerce
from sqlalchemy.dialects.postgresql import JSONB

class Resource(Base):
    __tablename__ = 'resources'
    id = Column(Integer, primary_key=True)
    data = Column(Text)
    
    @property
    def data_json(self):
        return type_coerce(self.data, JSONB)

# 调用查询
records = db_session.query(Resource).filter(
    Resource.data_json["lastname"] == "Doe"
).all()

MySQL #

通过cast函数将TEXT转为JSON类型，直接使用键索引语法：

from sqlalchemy import cast, JSON

records = db_session.query(Resource).filter(
    cast(Resource.data, JSON)["lastname"] == "Doe"
).all()

也可以用MySQL原生的json_extract函数：

from sqlalchemy import func

records = db_session.query(Resource).filter(
    func.json_extract(Resource.data, '$.lastname') == "Doe"
).all()

SQLite #

先启用SQLite的JSON扩展，再用json函数转换查询：

from sqlalchemy import func

# 启用JSON扩展
db_session.execute("SELECT json('{}');")

records = db_session.query(Resource).filter(
    func.json(Resource.data)["lastname"].astext == "Doe"
).all()

跨数据库通用方案 #

自定义TypeDecorator封装TEXT列，让它在代码层面表现得像原生JSON列，自动处理JSON字符串和Python对象的转换：

from sqlalchemy import Text, TypeDecorator
import json

class TextJSON(TypeDecorator):
    impl = Text

    def process_bind_param(self, value, dialect):
        return json.dumps(value) if value is not None else None

    def process_result_value(self, value, dialect):
        return json.loads(value) if value is not None else None

# 修改模型定义
class Resource(Base):
    __tablename__ = 'resources'
    id = Column(Integer, primary_key=True)
    data = Column(TextJSON)

# 现在可以直接用你想要的语法查询
records = db_session.query(Resource).filter(
    Resource.data["lastname"] == "Doe"
).all()

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