SQLAlchemy session.merge 抛出“无法更新标识列”

Question

我在数据库表中有条目，其中包含以下列：task_id、userstory_id、state、signed_to_mail
task_id 是主键
现在我正在尝试更新这些条目，我的意思是我再次收到任务，如果没有task_id，则将其添加到表中，但如果task_id已经存在，则只需更新更改的字段（task_id永远不会更改）
我正在尝试使用 session.merge() 执行此操作，然后尝试，但我正在执行 session.commit()
但我明白了

Cannot update identity column 'task_id'.

我发现 session.merge() 应该只更新更改的字段，而不应该引发身份异常。

def put_task(
    session: Any,
    parameters: dict[Any, Any],
) -> Any:
[...]
new_task = models.AzureTaskModel(
        task_id=parameters["task_id"],
        userstory_id=parameters["userstory_id"],
        state=parameters["state"],
        assigned_to_mail=parameters["assigned_to_mail"].lower()
    )
session.merge(instance=new_task)
    try:
        session.commit()
    except Exception as e:
        logging.warning(
            "Error during creating Task %s : %s",
            new_task.task_id,
            e,
        )
    return new_task

Answer 1

您看到的错误消息“无法更新身份列'task_id'”表明您正在尝试更新主键

task_id

，这是不允许的。数据库中表的主键用于唯一标识该表中的每条记录，因此一旦设置就无法更改。

您使用的

session.merge()

方法旨在仅更新已更改的字段，但如果“task_id”是标识列（自增主键），则无法更新。

您可能需要考虑使用不同的方法来更新记录，而不是使用

session.merge()

。首先，您可以检查该对象是否已存在于数据库中，如果存在，则更新必要的字段。

这是一个粗略的例子：

def put_task(
    session: Any,
    parameters: dict[Any, Any],
) -> Any:
[...]
new_task = session.query(models.AzureTaskModel).filter_by(task_id=parameters["task_id"]).first()
if new_task:
    # Update the necessary fields
    new_task.userstory_id=parameters["userstory_id"]
    new_task.state=parameters["state"]
    new_task.assigned_to_mail=parameters["assigned_to_mail"].lower()
else:
    # Create a new object and add it to the session
    new_task = models.AzureTaskModel(
        task_id=parameters["task_id"],
        userstory_id=parameters["userstory_id"],
        state=parameters["state"],
        assigned_to_mail=parameters["assigned_to_mail"].lower()
    )
    session.add(new_task)
try:
    session.commit()
except Exception as e:
    logging.warning(
        "Error during creating/updating Task %s : %s",
        new_task.task_id,
        e,
    )
return new_task