在FastAPI+SQLAlchemy中过滤PostgreSQL JSON数组列的问题

我用FastAPI结合SQLAlchemy操作PostgreSQL数据库，需要根据JSON数组列中的值过滤结果，但当前代码不仅过滤不符合预期，还触发了报错。

相关代码与数据 #

我的表中services列为JSON类型，简化后的SQLAlchemy模型：

class Element(Base):
    id = Column(Integer, primary_key=True, index=True)
    services = Column(JSON, nullable=True)

当前查询代码：

query = db.query(Element)

if filters:
    if 'services' in filters:
        query = query.filter(Element.services.contains([filters['services']]))

触发的报错信息：

FROM elements WHERE (elements.services LIKE '%%' || %(services_1)s || '%%')) AS anon_1] [parameters: {'services_1': '["Test"]'}] 500 Internal Server Error

services列的存储数据示例：

[
  {
    "id": 1,
    "services": ["service1", "service2", "Test"]
  },
  {
    "id": 2,
    "services": ["service3", "service4"]
  },
  {
    "id": 3,
    "services": ["service2", "service5", "Test"]
  }
]

期望用services="Test"过滤时，返回包含该值的记录，请问如何用SQLAlchemy和PostgreSQL正确实现？

解决方案 #

问题根源 #

你使用的contains方法在SQLAlchemy处理普通JSON列时，默认会生成基于字符串模糊匹配的LIKE语句，这和PostgreSQL的JSON数组存储逻辑不兼容，导致语法错误或匹配失效。PostgreSQL针对JSON类型有专门的查询语法，需要针对性调整。

正确实现方式 #

方法1：改用JSONB类型（推荐）

PostgreSQL的JSONB类型比普通JSON更适合查询场景，支持原生的包含判断操作符@>。

首先修改模型的列类型：

from sqlalchemy.dialects.postgresql import JSONB

class Element(Base):
    id = Column(Integer, primary_key=True, index=True)
    services = Column(JSONB, nullable=True)

然后调整查询代码：

query = db.query(Element)

if filters and 'services' in filters:
    target_service = filters['services']
    # 利用JSONB的@>操作符判断数组是否包含目标元素
    query = query.filter(Element.services.contains([target_service]))

方法2：保留JSON类型的原生函数写法

如果无法修改列类型，可以用PostgreSQL的json_array_elements函数展开数组后匹配，注意要去重：

from sqlalchemy import func

query = db.query(Element)

if filters and 'services' in filters:
    target_service = filters['services']
    query = query.filter(
        func.json_array_elements(Element.services).astext == target_service
    ).distinct()

效果说明 #

两种方法都能正确匹配到services数组中包含Test的记录（即id为1和3的条目），不会再触发语法错误。其中方法1的性能更优，因为JSONB类型支持索引优化。

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