SQLAlchemy 在一对一关系中替换子项时出错

Question

我想在用户和他们的最后一条消息之间建立一对一的关系。我希望当我更新他们的最后一条消息时，之前的最后一条消息将从数据库中删除。以下是最小的可重现示例

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)

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()

此操作失败并出现错误

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) NOT NULL constraint failed: message.from_user_id
[SQL: UPDATE message SET from_user_id=? WHERE message.id = ?]
[parameters: (None, '1')]
(Background on this error at: https://sqlalche.me/e/20/gkpj)

当我在

from_user_id

构造函数中显式提供它时，我不确定如何找不到

MessageEntity

。

Answer 1

这有效：

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

class Base(DeclarativeBase):
    id: Mapped[str] = mapped_column(primary_key=True)

class UserEntity(Base):
    __tablename__ = "user"

    last_message: Mapped["MessageEntity"] = relationship("MessageEntity", uselist=False)

class MessageEntity(Base):
    __tablename__ = "message"

    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")
    session.add(user1)
    session.add(user2)
    session.commit()
    user1.last_message = MessageEntity(id="1", content="hello", from_user_id="1")
    user1.last_message = MessageEntity(id="2", content="hello", from_user_id="1")
    session.add(user1)
    session.commit()