FastAPI+SQLAlchemy多对多关联表单条删除异常问题求助

基于FastAPI与SQLAlchemy构建应用时，在多对多关联的中间表device_protocol中，尝试删除device_id=1且protocol_id=2的单条记录，却意外删除了所有device_id=1的记录。以下是相关表结构、模型及路由代码，求实现单条记录的精准删除。

表结构示例 #

模型代码 #

class DeviceProtocolAssociation(Base):
    __tablename__ = "device_protocol"
    device_id = Column(Integer, ForeignKey("device.id", ondelete="CASCADE"), primary_key=True)
    device = relationship("Device", back_populates="device_protocols")
    protocol_id = Column(Integer, ForeignKey("protocol.id"), primary_key=True)
    protocol = relationship("Protocol", back_populates="device_protocols")
    status_id = Column(Integer, ForeignKey("status.id"), nullable=True)
    status = relationship("Status", back_populates="device_protocols")

class Device(Base):
    __tablename__ = "device"
    id = Column(Integer, primary_key=True, unique=True, index=True)
    name = Column(String(255))
    status_id = Column(Integer, ForeignKey('status.id'))
    status = relationship("Status", back_populates="devices")
    device_protocols = relationship(DeviceProtocolAssociation, back_populates="device")
    protocols = association_proxy("device_protocols", "protocols")

class Protocol(Base):
    __tablename__ = "protocol"
    id = Column(Integer, primary_key=True, unique=True, index=True)
    name = Column(String(255))
    device_protocols = relationship(DeviceProtocolAssociation, back_populates="protocol")
    devices = association_proxy("device_protocols", "devices")

class Status(Base):
    __tablename__ = "status"
    id = Column(Integer, primary_key=True, unique=True, index=True)
    name = Column(String(255))
    description = Column(String(255), nullable=True)
    devices = relationship("Device", back_populates="status")
    device_protocols = relationship(DeviceProtocolAssociation, back_populates="status")

原路由代码 #

@router.delete('/{device_id}/{protocol_id}')
async def delete_status(device_id: int, protocol_id: int, db:Session=Depends(get_db)):
    relation_query = db.query(DeviceProtocolAssociation).filter(DeviceProtocolAssociation.device_id==device_id and DeviceProtocolAssociation.protocol== protocol_id)
    db_relation = relation_query.first()
    if not db_relation:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f'No relation with this id: {id} found')
    relation_query.delete(db_relation)
    db.commit()
    return {"relation": "deleted"}

错误点总结 #

1. filter条件逻辑错误：使用Python原生and而非SQLAlchemy的and_（或链式调用filter），导致条件被解析为布尔值，最终仅生效device_id==device_id单一条件，误删所有该device_id的记录。
2. 字段引用错误：DeviceProtocolAssociation.protocol== protocol_id中，应引用字段protocol_id而非关系对象protocol。
3. 变量引用错误：异常信息中的{id}未定义，需改为device_id={device_id}, protocol_id={protocol_id}。
4. 删除方法调用错误：relation_query.delete(db_relation)不符合SQLAlchemy API规范，Query.delete()无需传入对象参数，或直接删除查询到的实例对象。

修正后的路由代码（两种实现方式） #

方式一：先查询实例再删除 #

from sqlalchemy import and_
from fastapi import HTTPException, status

@router.delete('/{device_id}/{protocol_id}')
async def delete_device_protocol_relation(device_id: int, protocol_id: int, db: Session = Depends(get_db)):
    db_relation = db.query(DeviceProtocolAssociation).filter(
        and_(
            DeviceProtocolAssociation.device_id == device_id,
            DeviceProtocolAssociation.protocol_id == protocol_id
        )
    ).first()
    
    if not db_relation:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'No relation found with device_id={device_id} and protocol_id={protocol_id}'
        )
    
    db.delete(db_relation)
    db.commit()
    return {"message": "Relation deleted successfully"}

方式二：直接通过条件删除（无需先查询实例） #

from sqlalchemy import and_
from fastapi import HTTPException, status

@router.delete('/{device_id}/{protocol_id}')
async def delete_device_protocol_relation(device_id: int, protocol_id: int, db: Session = Depends(get_db)):
    deleted_count = db.query(DeviceProtocolAssociation).filter(
        and_(
            DeviceProtocolAssociation.device_id == device_id,
            DeviceProtocolAssociation.protocol_id == protocol_id
        )
    ).delete()
    
    if deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f'No relation found with device_id={device_id} and protocol_id={protocol_id}'
        )
    
    db.commit()
    return {"message": f"Deleted {deleted_count} relation(s) successfully"}

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