如何在SQLAlchemy中仅匹配JSON第一层键实现应用名称精准搜索

模型定义 #

class City(Base):
    __tablename__ = 'Citys'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    metadatas = Column(String)

存储数据示例 #

id=1, name="London", metadatas='{"lat":51.5072, "lon":0.1276, "Mayor": "Sadiq Khan", "participating":true, "apps":{"Facebook": {"messenger": true}, "Instagram":{"message":false}}}',
id=2, name="Liverpool", metadatas='{"lat":53.4084, "lon":2.9916, "Mayor": "Joanne Anderson", "participating":false, "apps":{"Twitter": {"messenger": true}, "telegram":{"message":false}}}',
id=3, name="Manchester", metadatas='{"lat":53.4808, "lon":2.2426, "Mayor": "Donna Ludford", "participating":true, "apps":{"TikTok": {"messenger": true}, "Google Meet":{"message":false}}}'

需求 #

搜索应用名称时，返回匹配该名称的id、name和metadatas，仅匹配apps对象的一级键（比如搜“Facebook”返回伦敦数据），不匹配嵌套字段（搜“message”应返回空）。

当前实现问题 #

现有代码把apps转成字符串做模糊匹配，会误匹配嵌套字段：

filtered_query = session.query(Citys).filter(
    func.json_extract(Citys.metadatas, '$.apps').cast(String).like('%' + 'Facebook' + '%')).all()

搜“message”时会返回所有数据，不符合要求。

核心是直接检查JSON中apps的一级键是否包含目标值，而非转字符串模糊匹配，以下是主流数据库的实现方式：

1. PostgreSQL #

如果metadatas是JSONB类型（推荐），可以用jsonb_object_keys提取一级键后匹配；如果是字符串类型，先转成JSONB再处理：

from sqlalchemy import func, JSONB

target_app = "Facebook"
# 若metadatas是JSONB类型，去掉func.cast部分
filtered_query = session.query(City).filter(
    func.exists(
        session.query(func.jsonb_object_keys(func.cast(City.metadatas, JSONB)['apps'])).filter(
            func.jsonb_object_keys(func.cast(City.metadatas, JSONB)['apps']) == target_app
        )
    )
).all()

2. MySQL #

用JSON_KEYS提取apps的一级键列表，再用JSON_CONTAINS检查目标值是否在列表中：

from sqlalchemy import func

target_app = "Facebook"
filtered_query = session.query(City).filter(
    func.json_contains(
        func.json_keys(City.metadatas, '$.apps'),
        f'"{target_app}"'  # JSON格式的字符串需要带双引号
    )
).all()

3. SQLite #

需要启用json1扩展，用json_each遍历apps的键再匹配：

from sqlalchemy import func

target_app = "Facebook"
filtered_query = session.query(City).filter(
    func.exists(
        session.query(func.json_each(func.json_extract(City.metadatas, '$.apps'), '$.key')).filter(
            func.json_each(func.json_extract(City.metadatas, '$.apps'), '$.key') == target_app
        )
    )
).all()

通用优化建议 #

如果频繁做这类查询，建议把metadatas字段改成SQLAlchemy的JSON或JSONB类型（而非String），数据库处理JSON的效率更高，代码也更简洁：

from sqlalchemy.dialects.postgresql import JSONB
# 通用JSON类型：from sqlalchemy import JSON

class City(Base):
    __tablename__ = 'Citys'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    metadatas = Column(JSONB)  # 或JSON

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