如何使用SQLAlchemy和SQLite查询JSON中的列表？

问题场景 #

我在使用SQLAlchemy操作SQLite数据库时，遇到了JSON字段中存储列表的查询问题。我的模型定义和测试代码如下：

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

Session = sessionmaker()
Base = declarative_base()


class Track(Base):  # noqa: WPS230
    __tablename__ = "track"

    id = Column(Integer, primary_key=True)
    fields = Column(JSON(none_as_null=True), default="{}")

    def __init__(self, id):
        self.id = id
        self.fields = {}


engine = sqlalchemy.create_engine("sqlite:///:memory:")
Session.configure(bind=engine)
Base.metadata.create_all(engine)  # creates tables
session = Session()

track1 = Track(id=1)
track2 = Track(id=2)
track1.fields["list"] = ["wow"]
track2.fields["list"] = ["wow", "more", "items"]

session.add(track1)
session.add(track2)
session.commit()

我尝试过以下查询方式，但都无法得到预期结果：

• session.query(Track).filter(Track.fields["list"].as_string() == "wow").one()
• session.query(Track).filter(Track.fields["list"].as_string() == "[wow]").one()
• session.query(Track).filter(Track.fields["list"].as_json() == ["wow", "more", "items"]).one()

另外，用contains()方法会匹配元素的子字符串，这不是我想要的精确匹配效果。

解决方案 #

你用的是SQLite数据库，它的JSON处理依赖内置的JSON函数，得用SQLAlchemy的func调用这些函数才能实现精确匹配：

1. 精确匹配整个列表 #

要精准匹配某个完整列表，可以用json_extract提取JSON字段里的列表，再把目标列表转成JSON格式来比较：

from sqlalchemy import func

# 查找list为["wow"]的记录
result = session.query(Track).filter(
    func.json_extract(Track.fields, '$.list') == func.json(["wow"])
).one()
print(result.id)  # 输出1

# 查找list为["wow", "more", "items"]的记录
result2 = session.query(Track).filter(
    func.json_extract(Track.fields, '$.list') == func.json(["wow", "more", "items"])
).one()
print(result2.id)  # 输出2

2. 查询列表包含指定元素（精确匹配元素，非子字符串） #

如果要找列表里包含某个特定元素的记录，又不想匹配子字符串，就用json_contains函数，记得把目标元素转成JSON格式：

# 查找list中包含"more"的记录
result = session.query(Track).filter(
    func.json_contains(Track.fields, func.json("more"), '$.list')
).all()
print([r.id for r in result])  # 输出[2]

3. SQLAlchemy 1.4+的简洁写法 #

要是你用的是SQLAlchemy 1.4及以上版本，还能使用更简洁的语法：

from sqlalchemy import cast

# 精确匹配列表
result = session.query(Track).filter(
    Track.fields["list"].as_json() == cast(["wow"], JSON)
).one()

注意事项 #

• 确保你的SQLite版本在3.31.0及以上，该版本开始支持完整的JSON功能。
• 不同数据库的JSON查询语法有差异，如果后续切换到PostgreSQL等数据库，查询方式会有所不同，比如PostgreSQL支持直接用@>操作符检查包含关系。

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