调用含Pydantic模型的FastAPI接口时出现422错误求助

调用基于Pydantic模型的FastAPI接口时遇到422错误，移除请求体中d_f_values的布尔值后请求可成功，但接口会将d_f_values字段解析为None，以下是问题分析与解决方法：

涉及的Pydantic模型 #

class DishBase(BaseModel):
    d_name: constr(min_length=1)
    d_i_names: conlist(item_type=str, min_items=1)


class DishCreate(DishBase):
    d_i_weights: conlist(item_type=confloat(gt=0, lt=1), min_items=1)
    d_f_names: conlist(item_type=str, min_items=1)
    d_f_values: conlist(item_type=Union[bool, int, str, float], min_items=1)

    @validator("d_i_weights")
    def num_ingrs_must_match_num_weights(cls, v, values):
        if "d_i_names" in values and len(v) != len(values["d_i_names"]):
            raise ValueError("The number of ingredients and weights must match")
        return v

    @validator("d_i_weights")
    def weights_must_add_up_to_1(cls, v):
        if sum(v) != 1:
            raise ValueError("Weights must add up to 1")
        return v

    @validator("d_f_values")
    def num_feat_must_match_num_values(cls, v, values):
        if "d_f_names" in values and len(v) != len(values["d_f_names"]):
            raise ValueError("The number of features and values must match")
        return v

    @validator("d_f_values")
    def type_feat_values_must_match_with_db(cls, v, values):
        for index, feature_name in enumerate(values["d_f_names"]):
            conn, cursor = get_db()
            query = """SELECT * FROM features WHERE f_name = %s;"""
            value = (feature_name,)
            cursor.execute(query, value)
            row = dict(cursor.fetchone())
            cursor.close()
            conn.close()
            feature_type = row["f_type"]
            if type(v[index]) != feature_type:
                raise ValueError(f"type of feature {feature_name} is {feature_type}, but {type(v[index])} is provided")

FastAPI接口代码 #

@router.post("/", status_code=status.HTTP_201_CREATED, response_model=schemas.DishResponse)
def create_dish(dish: schemas.DishCreate):
    conn, cursor = get_db()
    query = """SELECT * FROM dishes WHERE d_name = %s;"""
    value = (dish.d_name,)
    cursor.execute(query, value)
    row = cursor.fetchone()
    cursor.close()
    conn.close()
    if row != None:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"This dish already exists")

    conn, cursor = get_db()
    query = """INSERT INTO dishes (d_name) VALUES (%s) RETURNING *;"""
    value = (dish.d_name,)
    cursor.execute(query, value)
    new_dish = dict(cursor.fetchone())
    conn.commit()
    cursor.close()
    conn.close()
    dish_id = new_dish["d_id"]

    for ingredient in dish.d_i_names:
        conn, cursor = get_db()
        query = """SELECT * FROM ingredients WHERE i_name = %s;"""
        value = (ingredient,)
        cursor.execute(query, value)
        row = cursor.fetchone()
        cursor.close()
        conn.close()
        if row == None:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"ingredient {ingredient} was not found")

    for index, ingredient in enumerate(dish.d_i_names):
        conn, cursor = get_db()
        query = """SELECT * FROM ingredients WHERE i_name = %s;"""
        value = (ingredient,)
        cursor.execute(query, value)
        row = dict(cursor.fetchone())
        cursor.close()
        conn.close()
        ingredient_id = row["i_id"]

        conn, cursor = get_db()
        query = """INSERT INTO dishes_ingredients (di_d_id, di_i_id, di_i_weight) VALUES (%s, %s, %s);"""
        values = (dish_id, ingredient_id, dish.d_i_weights[index])
        cursor.execute(query, values)
        conn.commit()
        cursor.close()
        conn.close()

    for feature in dish.d_f_names:
        conn, cursor = get_db()
        query = """SELECT * FROM features WHERE f_name = %s;"""
        value = (feature,)
        cursor.execute(query, value)
        row = cursor.fetchone()
        cursor.close()
        conn.close()
        if row == None:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"feature {feature} was not found")

    for index, feature in enumerate(dish.d_f_names):
        conn, cursor = get_db()
        query = """SELECT * FROM features WHERE f_name = %s;"""
        value = (feature,)
        cursor.execute(query, value)
        row = dict(cursor.fetchone())
        cursor.close()
        conn.close()
        feature_id = row["f_id"]

        conn, cursor = get_db()
        query = """INSERT INTO dishes_features (df_d_id, df_f_id, df_f_value) VALUES (%s, %s, %s);"""
        values = (dish_id, feature_id, str(dish.d_f_values[index]))
        cursor.execute(query, values)
        conn.commit()
        cursor.close()
        conn.close()

    return dish

请求体 #

{
  "d_name": "papa dish",
  "d_i_names": ["ingredient 1", "ingredient 2"],
  "d_i_weights": [0.3, 0.7],
  "d_f_names": ["feature A", "feature B", "feature C", "feature D"],
  "d_f_values": [True, 3, "fatty", 4.5]
}

错误详情 #

{
  "detail": [
    {
      "loc": [
        "body",
        191
      ],
      "msg": "Expecting value: line 6 column 18 (char 191)",
      "type": "value_error.jsondecode",
      "ctx": {
        "msg": "Expecting value",
        "doc": "{\n  \"d_name\": \"papa dish\",\n  \"d_i_names\": [\"ingredient 1\", \"ingredient 2\"],\n  \"d_i_weights\": [0.3, 0.7],\n  \"d_f_names\": [\"feature A\", \"feature B\", \"feature C\", \"feature D\"],\n  \"d_f_values\": [True, 3, \"fatty\", 4.5]\n}",
        "pos": 191,
        "lineno": 6,
        "colno": 18
      }
    }
  ]
}

问题原因 #

1. 初始422错误的原因 #

请求体中的布尔值使用了Python风格的大写True，但JSON规范要求布尔值必须是小写的true/false。FastAPI默认使用标准JSON解析器，无法识别大写的布尔值，直接触发JSON解码错误，返回422状态码。

2. 移除布尔值后d_f_values被解析为None的原因 #

Pydantic验证器type_feat_values_must_match_with_db存在逻辑错误：数据库中存储的f_type是类型的字符串标识（比如"bool"、"int"），但代码中直接将其与Python的类型对象（比如type(v[index])返回的<class 'int'>）做比较，两者永远不相等，导致验证失败。Pydantic在验证失败后无法正确解析该字段，最终将d_f_values设为None。

解决方案 #

1. 修复JSON格式错误 #

将请求体中的大写布尔值改为JSON规范的小写形式：

{
  "d_name": "papa dish",
  "d_i_names": ["ingredient 1", "ingredient 2"],
  "d_i_weights": [0.3, 0.7],
  "d_f_names": ["feature A", "feature B", "feature C", "feature D"],
  "d_f_values": [true, 3, "fatty", 4.5]
}

2. 修复验证器的类型比较逻辑 #

修改type_feat_values_must_match_with_db验证器，添加类型字符串到Python类型的映射，确保比较逻辑正确：

@validator("d_f_values")
def type_feat_values_must_match_with_db(cls, v, values):
    # 定义数据库类型字符串到Python类型的映射
    type_map = {
        "bool": bool,
        "int": int,
        "str": str,
        "float": float
    }
    # 优化数据库连接：避免循环内频繁开关连接
    conn, cursor = get_db()
    try:
        for index, feature_name in enumerate(values["d_f_names"]):
            query = """SELECT * FROM features WHERE f_name = %s;"""
            value = (feature_name,)
            cursor.execute(query, value)
            row = dict(cursor.fetchone())
            feature_type_str = row["f_type"]
            # 获取对应的Python类型
            expected_type = type_map.get(feature_type_str)
            if not expected_type:
                raise ValueError(f"Unsupported feature type: {feature_type_str}")
            # 检查传入值的类型是否匹配
            if not isinstance(v[index], expected_type):
                raise ValueError(f"Feature {feature_name} expects type {expected_type.__name__}, got {type(v[index]).__name__}")
    finally:
        cursor.close()
        conn.close()
    return v

额外优化建议 #

• 避免在循环内频繁打开/关闭数据库连接，应一次性打开连接，完成所有操作后再关闭，提升性能。
• 可以在验证器中添加pre=True参数，确保在Pydantic解析字段前先执行验证（如果需要提前处理值的话）。

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