SQLAlchemy一对一关系替换子对象时触发IntegrityError错误

问题场景 #

要建立用户与最后一条消息的一对一关系，预期更新用户最后一条消息时自动删除旧消息，但执行代码时触发NOT NULL constraint failed: message.from_user_id错误，尽管构造MessageEntity时已显式传入from_user_id。

错误原因 #

当替换user1.last_message时，SQLAlchemy默认会尝试解除旧消息与用户的关联：将旧消息的from_user_id设为NULL。但from_user_id字段被定义为非空（Mapped[str]默认带有NOT NULL约束），该操作违反数据库约束，导致报错。

解决方案 #

在UserEntity的last_message关系中添加cascade="all, delete-orphan"参数，同时确保双向关系的back_populates配置正确：

class UserEntity(Base):
    __tablename__ = "user"

    id: Mapped[str] = mapped_column(primary_key=True)
    last_message: Mapped["MessageEntity"] = relationship(
        "MessageEntity", 
        uselist=False, 
        back_populates="from_user",
        cascade="all, delete-orphan"
    )

修改后完整代码 #

from sqlalchemy import ForeignKey, create_engine
from sqlalchemy.orm import mapped_column, relationship, Mapped, DeclarativeBase

class Base(DeclarativeBase):
    pass

class UserEntity(Base):
    __tablename__ = "user"

    id: Mapped[str] = mapped_column(primary_key=True)
    last_message: Mapped["MessageEntity"] = relationship(
        "MessageEntity", 
        uselist=False, 
        back_populates="from_user",
        cascade="all, delete-orphan"
    )

class MessageEntity(Base):
    __tablename__ = "message"

    id: Mapped[str] = mapped_column(primary_key=True)
    content: Mapped[str] = mapped_column()
    from_user_id: Mapped[str] = mapped_column(ForeignKey('user.id'))
    from_user: Mapped["UserEntity"] = relationship("UserEntity", back_populates="last_message", foreign_keys=[from_user_id])


DEFAULT_SQL_URL = "sqlite+pysqlite:///:memory:"

engine = create_engine(DEFAULT_SQL_URL, echo=True)
Base.metadata.create_all(engine)

from sqlalchemy.orm import Session

with Session(engine) as session:
    user1 = UserEntity(id="1")
    user2 = UserEntity(id="2")
    user1.last_message = MessageEntity(id="1", content="hello", from_user_id="1")
    session.add(user1)
    session.add(user2)
    session.commit()
    user1.last_message = MessageEntity(id="2", content="hello", from_user_id="1")
    session.add(user1)
    session.commit()

关键说明 #

• cascade="all, delete-orphan"：all包含常用级联操作，delete-orphan表示当消息对象不再被用户关联时（成为“孤儿”），自动删除该消息，避免了修改外键为NULL的操作
• 补全back_populates确保双向关系同步，让SQLAlchemy正确处理对象间的关联状态

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