You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Flask应用中将嵌套列表作为POST接口参数传递

可行实现方案

Flask-RESTful自带的reqparse设计初衷是处理简单的查询参数、表单参数,本身对嵌套JSON结构的支持非常有限,针对你的场景有两种落地方式,可根据现有代码维护成本选择:


方案1:最小改动兼容现有reqparse逻辑

不需要替换现有解析器,只需要把列表、嵌套结构类型的参数作为整体注册到parser中,拿到原始值后自行做内部结构校验即可。

注意:action='append'仅适用于query/form参数重复传参的场景(比如?ci=xxx&ci=yyy),对于application/json类型的请求,直接指定参数名即可拿到对应字段的完整结构化值,不需要额外加append配置。

代码修改示例

首先补全parser的参数定义:

new_volume_parser = reqparse.RequestParser()
new_volume_parser.add_argument(
    "change_order",
    type=int,
    required=True,
    location="json"
)
new_volume_parser.add_argument(
    "environment",
    required=True,
    location="json",
    default="shared_env",
)
new_volume_parser.add_argument(
    "ci_list",
    type=list,
    required=True,
    location="json"
)
new_volume_parser.add_argument(
    "volumes",
    type=list,
    required=True,
    location="json"
)
new_volume_parser.add_argument(
    "dr_ci_list",
    type=list,
    required=False,
    location="json",
    default=[]
)

在接口方法中拿到参数后,自行校验volumes的内部结构:

@API.route("/flows/new_volume")
class NewVolume(SAFResource):
    @API.expect(new_volume_parser)
    def post(self):
        args = new_volume_parser.parse_args()
        volumes = args["volumes"]
        valid_repl_types = {"synchronous", "asynchronous"}

        # 遍历校验每个卷的结构
        for idx, vol in enumerate(volumes):
            # 校验元素类型
            if not isinstance(vol, dict):
                return {"msg": f"volumes第{idx+1}项格式错误,必须为对象类型"}, 400
            # 校验公共必填字段
            required_common = ["name", "size", "replicated"]
            for field in required_common:
                if field not in vol:
                    return {"msg": f"volumes第{idx+1}项缺少必填字段:{field}"}, 400
            # 校验字段类型
            if not isinstance(vol["name"], str) or not vol["name"].strip():
                return {"msg": f"volumes第{idx+1}项name必须为非空字符串"}, 400
            if not isinstance(vol["size"], int) or vol["size"] <= 0:
                return {"msg": f"volumes第{idx+1}项size必须为正整数"}, 400
            if not isinstance(vol["replicated"], bool):
                return {"msg": f"volumes第{idx+1}项replicated必须为布尔值"}, 400
            # 校验副本卷特殊规则
            if vol["replicated"]:
                if vol.get("repl_type") not in valid_repl_types:
                    return {"msg": f"volumes第{idx+1}项为副本卷,必须指定合法repl_type,可选值为synchronous/asynchronous"}, 400
            else:
                if "repl_type" in vol:
                    return {"msg": f"volumes第{idx+1}项为非副本卷,无需传入repl_type"}, 400
        
        # 校验通过后执行后续业务逻辑
        # ...

这个方案的优点是完全兼容现有代码结构,改动量极小;缺点是嵌套结构的校验逻辑需要手动编写,字段较多时维护成本高,容易遗漏校验规则。


方案2:用结构化校验库替换reqparse(长期维护推荐)

reqparse官方早已停止功能迭代,对于复杂嵌套JSON请求,更推荐直接使用专门的参数校验库(比如pydantic、marshmallow),这类库天生支持任意层级的嵌套结构定义,自动完成类型转换、规则校验,代码可维护性更高。
以pydantic为例,实现步骤如下:

  1. 定义嵌套的请求体模型,所有字段规则、校验逻辑都在模型中声明
from pydantic import BaseModel, Field, root_validator, ValidationError
from typing import List, Literal, Optional
from flask import request

# 定义单个卷的结构模型
class Volume(BaseModel):
    name: str = Field(min_length=1, description="卷名称")
    size: int = Field(gt=0, description="卷大小,正整数")
    replicated: bool = Field(description="是否为副本卷")
    repl_type: Optional[Literal["synchronous", "asynchronous"]] = Field(None, description="副本同步模式,副本卷必填")

    @root_validator
    def check_repl_rule(cls, values):
        # 自定义校验规则:副本卷必须传repl_type,非副本卷不能传
        if values.get("replicated") and not values.get("repl_type"):
            raise ValueError("副本卷必须指定repl_type字段")
        if not values.get("replicated") and values.get("repl_type"):
            raise ValueError("非副本卷无需传入repl_type字段")
        return values

# 定义完整请求体模型
class NewVolumeReq(BaseModel):
    change_order: int = Field(description="变更单号")
    environment: str = Field(default="shared_env", description="所属环境")
    ci_list: List[str] = Field(min_length=1, description="主机CI列表")
    volumes: List[Volume] = Field(min_length=1, description="待创建卷列表")
    dr_ci_list: List[str] = Field(default=[], description="DR主机CI列表")
  1. 简化接口逻辑,直接从请求中拿到JSON后用模型校验
@API.route("/flows/new_volume")
class NewVolume(SAFResource):
    def post(self):
        try:
            # 自动校验、类型转换
            req_data = NewVolumeReq(**request.get_json())
        except ValidationError as e:
            return {"msg": "参数校验失败", "detail": e.errors()}, 400
        
        # 校验通过后直接使用结构化数据,所有字段类型、规则都符合预期
        # 比如req_data.volumes就是校验完成的卷列表,req_data.ci_list是字符串列表
        # 后续业务逻辑直接写即可
        # ...

这个方案的优点是结构清晰,不管嵌套多少层复杂结构都能轻松支持,不需要手写大量重复的类型判断逻辑,错误信息自动生成,后续迭代维护成本极低。


额外注意

你示例中的curl命令缺少JSON请求头,会导致Flask无法正确解析结构化请求体,正确的调用方式如下:

curl -X POST http://localhost/flows/new_volume \
-H "Content-Type: application/json" \
-d '{
    "change_order": 123456,             
    "environment": "stor env",   
    "ci_list": ["CI12345678", "HI12345"],                    
    "volumes": [
        {"name": "volume1", "size": 10, "replicated": false},
        {"name": "volume2", "size": 500, "replicated": true, "repl_type": "synchronous"}
    ],
    "dr_ci_list": ["CI00065232"]
}'

内容的提问来源于stack exchange,提问作者Student

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.26 11:51:20