属性错误：“元组”对象没有属性“_bulk_update_tuples”

Question

我正在使用flask和falsk-sqlalchemy开发Restful API，并尝试更新投诉表并更改其assignee_id和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'

下面是 API 的路由器部分

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

下面是API的控制器部分

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

这是我的投诉模型

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})"

我希望查询能够正常工作，并且 assingee_id 和 status_id 应该在数据库上更新

Answer 1

我相信正确的查询应该是：

query = update(Complaint).where(Complaint.c.id==complaint_id).values(
    assignee_id=user_id, 
    complaint_status_id=status_id
)

请注意，我在

.c

和

Complaint

之间添加了

.id

。

我还没有测试过它，但从错误来看，sqlalchemy 似乎正在使用元组而不是

QueryableAttribute

实例来处理，这是错误报告丢失的具有

_bulk_update_tuples

属性的类。

我还从 SQLAlchemy 的文档中找到了其他示例，这些示例指出所有 WHERE 表达式（及相关）都可以通过访问

.c

属性来完成。