GCP Spanner JSON列插入/更新随机报错排查求助

问题描述 #

使用SQLAlchemy ORM操作GCP Spanner数据库时，向JSON列插入/更新数据会随机触发以下报错：

google.api_core.exceptions.InvalidArgument: 400 Invalid value for bind parameter a1: Expected JSON.

本地Spanner模拟器可正常执行所有操作，云端失败的相同数据在本地也能成功。手动用json.dumps()序列化数据后依然报错。

相关代码与结构 #

表定义：

from sqlalchemy import Column, String, JSON, Float
from models.base import Base

class UtterData(Base):
    __tablename__ = "tbl_utter_data"

    conv_id = Column(String, primary_key=True)
    msg_id = Column(String, primary_key=True)
    key_indicator = Column(JSON)
    med_indicator = Column(JSON)

插入/更新逻辑：

from models.utter_data import UtterData
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

engine = create_engine("my_connection_url")
session = Session(engine)

data_obj = {"conv_id":"convID","msg_id":"messageID","key_indicator":[{"value": "None", "KeyID": "key0"}], "med_indicator": [{"value": "None", "MedID": "med7"}]}

# 插入操作
record = UtterData(data_obj)
session.add(record)
session.commit()

# 更新操作
fetched_record = session.query(UtterData).filter_by(
        conv_id='convID', msg_id='messageID'
)
fetched_record.update({"key_indicator":data_obj['key_indicator'], "med_indicator":data_obj['med_indicator']})

SQLAlchemy生成的报错查询示例：

sqlalchemy.exc.ProgrammingError: (google.cloud.spanner_dbapi.exceptions.ProgrammingError) []
[SQL: UPDATE tbl_utter_data SET key_indicator=%s, med_indicator=%s WHERE tbl_utter_data.conv_id = %s AND tbl_utter_data.msg_id = %s]
[parameters: [[{"value": "None", "KeyID": "key0"}], [{"value": "None", "MedID": "med7"}],'convID', 'messageID']]

可能的解决方法 #

1. 使用Spanner专属JSON类型替换SQLAlchemy默认JSON #

SQLAlchemy默认JSON类型可能在云端Spanner的适配场景下存在序列化异常，改用Spanner SQLAlchemy扩展提供的专属类型：

# 替换原JSON导入
from google.cloud.sqlalchemy_spanner import JSON as SpannerJSON

class UtterData(Base):
    __tablename__ = "tbl_utter_data"
    conv_id = Column(String, primary_key=True)
    msg_id = Column(String, primary_key=True)
    key_indicator = Column(SpannerJSON)
    med_indicator = Column(SpannerJSON)

2. 避免批量更新，改用实例属性修改方式 #

从报错参数可见JSON数组被额外嵌套了一层（[[...]]），这是批量更新的参数包装问题。改用实例直接修改的方式：

# 替换原更新逻辑
fetched_record = session.query(UtterData).filter_by(
    conv_id='convID', msg_id='messageID'
).first()
if fetched_record:
    fetched_record.key_indicator = data_obj['key_indicator']
    fetched_record.med_indicator = data_obj['med_indicator']
    session.commit()

3. 确保会话的正确生命周期管理 #

随机报错可能和会话复用、状态不一致有关，使用上下文管理器确保会话正确创建与销毁：

# 用上下文管理器管理会话
with Session(engine) as session:
    # 插入操作
    record = UtterData(data_obj)
    session.add(record)
    session.commit()

    # 更新操作
    fetched_record = session.query(UtterData).filter_by(
        conv_id='convID', msg_id='messageID'
    ).first()
    if fetched_record:
        fetched_record.key_indicator = data_obj['key_indicator']
        fetched_record.med_indicator = data_obj['med_indicator']
        session.commit()

4. 升级依赖到兼容版本 #

检查并升级相关依赖包，确保版本匹配：

pip install --upgrade google-cloud-spanner sqlalchemy python-spanner-sqlalchemy

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