在SQLAlchemy中，我应该如何指定关系字段是必需的?

Question

我有一个模型，它依赖于另一个模型的一些字段。这些字段应该在创建记录时出现，但我没有看到在数据库上强制执行的方法:

class Study(db.Model):
id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
type = db.Column(Enum(StudyTypeChoices), nullable=False)
owner_id = db.Column(UUID(as_uuid=True), db.ForeignKey('owner.id'), nullable=False)
participants = db.relationship('Participant', lazy=True, cascade='save-update, merge, delete')

我如何确保在创建研究记录时提供"参与者"(类似于"类型"字段发生的情况)?我知道我可以在它周围放一个包装器来确保这一点，但我想知道是否有一种更简洁的方法来使用sqlalchemy。

编辑:这是Participant模型的定义

class Participant(UserBase):
id = db.Column(UUID(as_uuid=True), db.ForeignKey("user_base.id"), primary_key=True)
study_id = db.Column(UUID(as_uuid=True), db.ForeignKey('study.id'))

Answer 1

您可以侦听before_flush事件，并通过引发异常来防止包含没有参与者的研究的刷新。

@event.listens_for(Session, "before_flush")
def before_flush(session, flush_context, instances):
for instance in session.new:  # might want to inspect session.dirty as well
if isinstance(instance, Study) and (
instance.participants is None or instance.participants == []
):
raise ValueError(
f"Study {instance} cannot have {instance.participants} participants."
)

这只检查新的研究，你可能想检查session.dirty以及更新的研究。

完整的演示:

from sqlalchemy import Column, ForeignKey, Integer, create_engine, event
from sqlalchemy.orm import Session, declarative_base, relationship
Base = declarative_base()

class Study(Base):
__tablename__ = "study"
id = Column(Integer, primary_key=True)
participants = relationship("Participant", uselist=True, back_populates="study")

class Participant(Base):
__tablename__ = "participant"
id = Column(Integer, primary_key=True)
study_id = Column(Integer, ForeignKey("study.id"), nullable=True)
study = relationship("Study", back_populates="participants")

@event.listens_for(Session, "before_flush")
def before_flush(session, flush_context, instances):
for instance in session.new:  # might want to inspect session.dirty as well
if isinstance(instance, Study) and (
instance.participants is None or instance.participants == []
):
raise ValueError(
f"Study {instance} cannot have {instance.participants} participants."
)
engine = create_engine("sqlite://", future=True, echo=True)
Base.metadata.create_all(engine)
s1 = Study()
p1_1 = Participant()
p1_2 = Participant()
s1.participants.extend([p1_1, p1_2])
s2 = Study()
with Session(bind=engine) as session:
session.add(s1)
session.commit()  # OK
with Session(bind=engine) as session:
session.add(s2)
session.commit()  # ValueError