在 SQLAlchemy 中检查外键的外键

Question

我有这 3 个 sql alchemy (sqla) 模型：

class Site(Base):
    __tablename__ = "site"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow())
    updated_at = Column(DateTime, nullable=True, default=datetime.utcnow(), onupdate=datetime.utcnow)


class Camera(Base):
    __tablename__ = "camera"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    site_id = Column(UUID(as_uuid=True), ForeignKey("site.id"), nullable=False)
    name = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow())
    updated_at = Column(DateTime, nullable=True, default=datetime.utcnow(), onupdate=datetime.utcnow)
    site = relationship("Site", backref="cameras")


class RtspServerEndpoint(Base):
    __tablename__ = "rtsp_server_endpoint"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    camera_id = Column(UUID(as_uuid=True), ForeignKey("camera.id"), nullable=False)
    rtsp_url_endpoint = Column(String, nullable=False)
    rtsp_username = Column(String, nullable=False)
    rtsp_encrypted_password = Column(String, nullable=False)
    name = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow())
    updated_at = Column(DateTime, nullable=True, default=datetime.utcnow(), onupdate=datetime.utcnow)
    camera = relationship("Camera", backref="rtsp_server_endpoint", lazy="joined")

camera_id是rtspserverendpoint表的外键，site_id是Camera表的外键。

当用户想要添加新的rtspserverendpoint 记录时，他会发出如下 HTTP 请求： POST 站点//camera//rtspserverendpoint

在添加新的rtspserverendpoint 之前，我想确保<site_id> 和<camera_id> 是一致的，作为安全措施。我可能可以进行单独的查询来检查，例如：

check_record_exist = db.session.query(Camera).filter(Camera.site_id == site_id).first()
if not check_record_exist:
    raise ("No such camera with this site_id")

但我想知道的是，是否有更优雅的方法来检查：例如，在我的 Base 模型中添加一个约束，禁止在数据库中添加这种不一致的记录。

Answer 1

我不知道有什么直接的方法可以直接在数据库上实现这种 2 级检查。

事实上，数据库应该知道的唯一一致性是您的新RtspServerEndpoint 实例将属于正确的Camera 实例。但默认情况下这将是正确的，因为您将创建 RtspServerEndpoint 实例。

因此，在我看来，检查POST请求的URL中site_id的正确性应该在你的代码逻辑中实现。我可能会这样做：

@handler(..., method='POST')
def rtspserverendpoint(site_id: int, camera_id: int):
    # find camera, which will allow us to check the correctness of the site_id as well
    camera = db.session.query(Camera).get(camera_id)

    if camera is None:
        raise Exception(f"No camera with this {camera_id=}.")

    if camera.site_id != site_id:
        raise Exception(f"Camera with this {camera_id=} does not belong to the site with {site_id=}.")

    new_obj = RtspServerEndpoint(
        ...,
        camera_id=camera_id,
        ...,
    )

    db.session.add(new_obj)
    db.session.commit()