如何在FastAPI中通过SQLAlchemy筛选JSON字段数据

原始代码场景 #

你给出的模型与接口代码如下：

from sqlalchemy import Column, JSON
from sqlalchemy.ext.declarative import declarative_base
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

Base = declarative_base()
router = APIRouter()

class Database_Model(Base):
    __tablename__ = "database_model"
    data = Column(JSON, nullable=False)

@router.get("/test/")
def test(
    db: Session = Depends(get_db)
):
    return (
        db.query(Database_Model)
            .filter(Database_Model.data["test_field"] == "value").all()
    )

直接用Database_Model.data["test_field"] == "value"无法适配所有数据库，需根据你使用的数据库类型调整筛选逻辑：

针对PostgreSQL的写法 #

PostgreSQL原生支持JSON/JSONB类型，SQLAlchemy 1.4+可直接通过astext转换匹配，也可用数据库原生函数：

from sqlalchemy import func

@router.get("/test/")
def test(db: Session = Depends(get_db)):
    # 方式1：直接转换JSON字段为文本匹配
    results = db.query(Database_Model).filter(
        Database_Model.data["test_field"].astext == "value"
    ).all()

    # 方式2：使用JSON路径匹配函数
    results = db.query(Database_Model).filter(
        func.jsonb_path_match(Database_Model.data, '$.test_field == "value"')
    ).all()
    
    return results

针对MySQL的写法 #

MySQL需调用JSON_EXTRACT函数提取JSON字段值：

from sqlalchemy import func

@router.get("/test/")
def test(db: Session = Depends(get_db)):
    results = db.query(Database_Model).filter(
        func.json_extract(Database_Model.data, "$.test_field") == "value"
    ).all()
    return results

跨数据库通用写法 #

如果需要兼容多种数据库，可通过cast将JSON字段转为字符串后匹配：

from sqlalchemy import cast, String

@router.get("/test/")
def test(db: Session = Depends(get_db)):
    results = db.query(Database_Model).filter(
        cast(Database_Model.data["test_field"], String) == "value"
    ).all()
    return results

注意：确保数据库中data字段的JSON结构确实包含test_field键，否则筛选结果会为空。

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