FastAPI 422错误排查：批量更新多行数据失败求助

问题描述 #

需要更新values表中符合customer_id和产品分组条件的所有数据行，通过join筛选指定分组下的产品。相关代码如下：

数据表模型 #

class Values(Base):
    __tablename__ = 'values'
    
    customer_id = Column(Integer, ForeignKey('customer.customer_id'), primary_key=True, nullable=False) 
    product_id = Column(Integer, ForeignKey('product.product_id'), primary_key = True, nullable= False) 
    quantity = Column(Numeric) 

接口路由 #

@router.put('/{customer_id}/{group}')
def update_values(customer_id: int, group:str, updated: schemas.Quantity, db: Session = Depends(get_db)):  
    sub_query = db.query(models.Values, models.Product.group).join(models.Product, models.Product.product_id == models.Values.product_id).filter(models.Values.customer_id==customer_id, models.Product.group==group).subquery()
    main_query = db.query(models.Values).filter(models.Values.product_id.in_(sub_query))
    
    main_query.update({models.Values.quantity: updated.item_value}, synchronize_session=False)
    db.commit()
    return main_query.all()

请求Schema #

class Quantity(BaseModel):
    item_value: float

调用接口时触发422错误，错误信息：

{
  "detail": [
    {
      "loc": [
        "path",
        "product_id"
      ],
      "msg": "value is not a valid integer",
      "type": "type_error.integer"
    }
  ]
}

错误原因分析 #

1. 请求URL与路由定义不匹配：
   错误信息明确指出路径参数product_id不是有效整数，但当前接口路由定义的是/{customer_id}/{group}，根本不存在product_id参数。这说明实际请求的URL路径不符合该接口的路由规则，比如：

   • 误调用了其他包含product_id参数的接口且传入了非整数值
   • 调用当前接口时参数顺序错误、路径格式写错，导致FastAPI误将某个参数识别为product_id

2. 数据库查询逻辑存在问题：
   即使解决了请求参数的问题，当前代码的查询逻辑也会报错。sub_query查询的是models.Values对象和models.Product.group字段，后续用product_id.in_(sub_query)时，子查询返回的是(Values实例, 分组名)的元组集合，而非单独的product_id列表，无法正确匹配。

修复方案 #

1. 修正请求URL #

确保请求的URL严格匹配接口路由/{customer_id}/{group}：

• customer_id传入整数（比如/1001/electronics）
• 不要在路径中加入额外的product_id参数
• 检查是否误调用了其他接口

2. 修复数据库查询逻辑 #

修改子查询，只返回需要的product_id字段，同时更新后重新查询获取最新数据：

@router.put('/{customer_id}/{group}')
def update_values(customer_id: int, group:str, updated: schemas.Quantity, db: Session = Depends(get_db)):  
    # 子查询仅查询符合条件的product_id
    sub_query = db.query(models.Values.product_id).join(models.Product, models.Product.product_id == models.Values.product_id).filter(models.Values.customer_id==customer_id, models.Product.group==group).subquery()
    main_query = db.query(models.Values).filter(models.Values.product_id.in_(sub_query))
    
    main_query.update({models.Values.quantity: updated.item_value}, synchronize_session=False)
    db.commit()
    # 更新后重新查询，确保返回最新数据
    return db.query(models.Values).filter(models.Values.product_id.in_(sub_query)).all()

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