Flask-SQLAlchemy更新报错：tuple对象无_bulk_update_tuples属性

使用Flask和Flask-SQLAlchemy开发RESTful API，尝试更新Complaint表的assignee_id和complaint_status_id字段时，出现如下错误：

127.0.0.1 - - [21/Dec/2023 15:28:37] "PATCH /complaint/1/assign HTTP/1.1" 500 -
Traceback (most recent call last):
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask\app.py", line 1478, in __call__
    return self.wsgi_app(environ, start_response)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask\app.py", line 1458, in wsgi_app
    response = self.handle_exception(e)
               ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask\app.py", line 1455, in wsgi_app
    response = self.full_dispatch_request()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask\app.py", line 869, 
in full_dispatch_request
    rv = self.handle_user_exception(e)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask\app.py", line 867, 
in full_dispatch_request
    rv = self.dispatch_request()
         ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask\app.py", line 852, 
in dispatch_request
    return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\flask_jwt_extended\view_decorators.py", line 170, in decorator
    return current_app.ensure_sync(fn)(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\routes\complaints\assign_complaint.py", line 25, in assign_complaint_route
    isUpdated = AssignComplaint.assign_complaint_controller(
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\controllers\complaints\assign_complaint.py", line 17, in assign_complaint_controller
    query = update(Complaint).where(Complaint.id==complaint_id).values(
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<string>", line 2, in values
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\sqlalchemy\sql\base.py", 
line 283, in _generative
    x = fn(self, *args, **kw)
        ^^^^^^^^^^^^^^^^^^^^^
  File "<string>", line 2, in values
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\sqlalchemy\sql\base.py", 
line 316, in check
    return fn(self, *args, **kw)
           ^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\sqlalchemy\sql\dml.py", line 1157, in values
    coerced_arg = dict(kv_generator(self, arg.items(), True))
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\sqlalchemy\orm\bulk_persistence.py", line 449, in _get_crud_kv_pairs
    return list(
           ^^^^^
  File "C:\Users\huzai\Documents\flask_projects\flask_with_sqlalchemy\venv\Lib\site-packages\sqlalchemy\orm\bulk_persistence.py", line 392, in _get_orm_crud_kv_pairs
    desc._bulk_update_tuples(v),
AttributeError: 'tuple' object has no attribute '_bulk_update_tuples'

路由代码 #

from flask import Blueprint, request
from flask_jwt_extended import jwt_required, get_jwt_identity
from services.response_services import display_response
from controllers.complaints import assign_complaint as AssignComplaint

assign_complaint_bp = Blueprint("assign-complaint", "complaint_services")
allowed_roles =["superadmin", "admin"]

@assign_complaint_bp.route("/complaint/<comp_id>/assign", methods=["PATCH"])
@jwt_required()
def assign_complaint_route(comp_id):
    user_data = get_jwt_identity()
    if (user_data.get("role") not in allowed_roles):
        return display_response(status_code=400, errors=["Unauthenticated user"])
    if not request.is_json:
        return display_response(
            errors=[
                "API accepts JSON data"
            ],
            status_code=400
        )
    data = request.get_json()
    if len((errors := validate_data(data))) > 0:
        return display_response(errors=errors, status_code=400)
    isUpdated = AssignComplaint.assign_complaint_controller(
        complaint_id=comp_id,
        user_id=data.get("staff_id")
    )
    if (isUpdated):
        return display_response(message="Complaint assigned successfully")
    else:
        return display_response(status_code=400, errors=["No update detect in given staff_id or Invalid complaint id or Invalid staff id"])

def validate_data(data):
    error_msgs = []
    if (data.get("staff_id") is None) or (len(str(data.get("staff_id")).strip()) == 0):
        error_msgs.append("staff_id is required")
    return error_msgs

控制器代码 #

from models.setup import db
from models.models.user_model import User
from models.models.complaint_status_model import ComplaintStatus
from models.models.complaint_model import Complaint
from sqlalchemy import select, update

def assign_complaint_controller(complaint_id, user_id):
    user_id = db.session.execute(select(User.id).where(
        User.role_type=="staff",
        User.id==user_id,
        User.is_active=="1",
    )).scalar()
    if (user_id is None): return False
    status_query = select(ComplaintStatus.id).where(ComplaintStatus.name=="ASSIGNED")
    status_id = db.session.execute(status_query).scalar()
    print(user_id, status_id, complaint_id)
    query = update(Complaint).where(Complaint.id==complaint_id).values(
            assignee_id=user_id, 
            complaint_status_id=status_id
            )
    result = db.session.execute(query)
    db.session.commit()
    return result.rowcount > 0

Complaint模型代码 #

from models.setup import db
import sqlalchemy
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime

class Complaint(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(sqlalchemy.String(45), nullable=False)
    description: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False)
    complaint_type_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey("complaint_type.id"))
    complaint_status_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey("complaint_status.id"))
    assignee_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey("user.id"), nullable=True),
    file_url: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(sqlalchemy.DateTime(timezone=True), server_default=sqlalchemy.sql.func.now())

    def __repr__(self) -> str:
        return f"Complaint(id={self.id}, title={self.title})"

预期结果：查询正常执行，assignee_id和status_id成功更新到数据库。

错误根源是Complaint模型中assignee_id字段定义末尾多了一个逗号，这会导致该字段被解析为元组而非SQLAlchemy的Mapped类型，批量更新时SQLAlchemy无法识别字段类型，从而抛出AttributeError。

修改模型代码，删除assignee_id行末尾的逗号：

# 修改前
assignee_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey("user.id"), nullable=True),

# 修改后
assignee_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey("user.id"), nullable=True)

修改完成后重新运行API，更新操作即可正常执行。

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