显式设置外键列后提交触发NotNullViolation错误排查

保存代码 #

def save_reply(self, reply: PostgresReplyTable) -> PostgresReplyTable:
    """
    Save a reply

    Args:
        job: The job to save the reply for
        text: The text of the reply

    Returns:
        PostgresReplyTable: The reply
    """
    reply = self._session.merge(reply)
    self._session.commit()
    return reply

调试信息 #

提交前调试显示used_prompt_id字段已有值：

/postgres_connector.py(444)save_reply()
-> self._session.commit()
(Pdb) reply
<database_sdk.models.reply.PostgresReplyTable object at 0x7fd3d0179240>
(Pdb) reply.used_prompt_id
2

提交错误信息 #

执行commit()触发非空约束错误：

(Pdb) n
sqlalchemy.exc.IntegrityError: (psycopg2.errors.NotNullViolation) null value in column "used_prompt_id" of relation "reply" violates not-null constraint
DETAIL:  Failing row contains (1, This is a reply to the job based on the use case requirements., d4e8e3e7-8a8c-4f2e-8e5e-0e2a7a3c9b0d, 2, null, null, null, null, 2024-06-10 11:59:40.913716, 2024-06-10 11:59:40.913724, 2024-06-10 11:59:40.254929).

[SQL: UPDATE reply SET used_prompt_id=%(used_prompt_id)s WHERE reply.reply_id = %(reply_reply_id)s]
[parameters: {'used_prompt_id': None, 'reply_reply_id': 1}]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
2

模型定义 #

class PostgresReplyTable(Base):
    __tablename__ = "reply"
    reply_id: Mapped[int] = mapped_column(Integer, primary_key=True)
    text: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    job_id: Mapped[str] = mapped_column(ForeignKey("job.job_id"))
    use_case_id: Mapped[int] = mapped_column(ForeignKey("use_case.id"))
    used_prompt_id: Mapped[int] = mapped_column(ForeignKey("prompt.prompt_id"))
    feedback_rate: Mapped[Optional[int]]
    feedback_comment: Mapped[Optional[str]] = mapped_column(Text)
    input_text: Mapped[Optional[str]] = mapped_column(Text)
    job: Mapped["PostgresJobTable"] = relationship(back_populates="reply")  # type: ignore
    use_case: Mapped["PostgresUseCaseTable"] = relationship(back_populates="replies")  # type: ignore
    used_prompt: Mapped["PostgresPromptTable"] = relationship(back_populates="reply")
    processing_start_date: Mapped[Optional[datetime]]
    processing_end_date: Mapped[Optional[datetime]]
    created_at: Mapped[datetime] = mapped_column(default=datetime.now())

解决方法 #

1. 调整merge操作的对象处理逻辑
   若传入的reply是游离态对象，merge返回的新对象可能未同步used_prompt_id值：

   • 若是新增对象，改用add替代merge：

     self._session.add(reply)
     self._session.commit()
     

   • 若是更新对象，merge后先执行flush同步内存状态：

     reply = self._session.merge(reply)
     self._session.flush()
     self._session.commit()
     

2. 补全模型字段的mapped_column定义
   模型中feedback_rate、processing_start_date、processing_end_date未显式指定mapped_column，可能导致ORM映射异常，补全定义：

   feedback_rate: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
   processing_start_date: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
   processing_end_date: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True)
   

3. 手动同步会话内对象状态
   直接查询数据库中已有对象，手动更新字段后提交，避免merge的状态不一致问题：

   existing_reply = self._session.get(PostgresReplyTable, reply.reply_id)
   if existing_reply:
       existing_reply.used_prompt_id = reply.used_prompt_id
       # 同步其他需要更新的字段
       self._session.commit()
       return existing_reply
   else:
       self._session.add(reply)
       self._session.commit()
       return reply
   

4. 验证模型与数据库schema一致性
   确认数据库中reply表的used_prompt_id字段确实是非空约束，且模型定义与数据库schema完全匹配。若模型修改过但未迁移数据库，需执行schema迁移操作。

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