sqlalchemy 多对多字段，其中额外字段未更新

Question

我有两个表用户和电影。这些表与额外的字段角色具有多对多的关系。如果用户和电影匹配，我正在尝试更新角色。但这给了我错误

user_film_table = Table(
  "user_film_association",
  Base.metadata,
  Column("id", Integer, primary_key=True, index=True),
  Column("user_id", ForeignKey("users.id"), index=True),
  Column("film_id", ForeignKey("film.id"), index=True),
  Column('role', ENUM(UserRoleEnum))  # Role can be 'writer', 'producer', or 'director' 
)

def update_user_role(self, user_id, film_id, new_role):
        user = self.db.query(User).filter(User.id == user_id).first()
        if not user:
            return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

        film = self.db.query(Film).filter(Film.id == film_id).first()
        if not film:          
            return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Film not found")

        if film not in user.film:
            return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User is not associated with the specified film")

        user_film = self.db.query(user_film_table).filter(user_film_table.c.user_id == user_id, user_film_table.c.film_id == film_id).first()
        print(user_film[0])
        user_film[3].role = new_role.value   #facing error while updating the role
        self.db.add(user_film)
        self.db.commit()
        return {"message": f"User role in film updated to {new_role} successfully", 'status_code': status.HTTP_202_ACCEPTED}

无法更新多对多字段中的角色 出现错误

raise exc.UnmappedInstanceError(instance) from err
sqlalchemy.orm.exc.UnmappedInstanceError: Class 'sqlalchemy.engine.row.Row' is not mapped

Answer 1

您遇到的错误是因为您尝试使用

user_film

添加的

self.db.add(user_film)

对象不是映射实例，而是行结果。

在 SQLAlchemy 中，只能将“映射”实例（已映射到数据库表的类实例）添加到会话中以进行 ORM 样式的 CRUD 操作。

您可以使用SQLAlchemy提供的

update

语句，该语句可用于更新一行中的特定字段。

where

方法用于指定应更新哪些行，

values

方法用于指定字段的新值。 以下是解决此问题的方法：

from sqlalchemy import update

def update_user_role(self, user_id, film_id, new_role):
    user = self.db.query(User).filter(User.id == user_id).first()
    if not user:
        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    film = self.db.query(Film).filter(Film.id == film_id).first()
    if not film:          
        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Film not found")

    if film not in user.film:
        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User is not associated with the specified film")

    # Use the update statement
    stmt = (
        update(user_film_table).
        where(user_film_table.c.user_id == user_id, user_film_table.c.film_id == film_id).
        values(role = new_role.value)
    )
    self.db.execute(stmt)
    self.db.commit()
    return {"message": f"User role in film updated to {new_role} successfully", 'status_code': status.HTTP_202_ACCEPTED}