如何在 FastAPI 中将字典列表作为 Body 参数发送？

Question

在 FastAPI 中传递字典列表，一般我们会定义一个 pydantic 模式，并会提到

param: List[schema_model]

我面临的问题是我有文件要附加到我的请求中。我找不到在路由器功能中定义架构和文件上传的方法。为此，我将所有参数（请求正文）定义为正文参数，如下所示。

@router.post("/", response_model=DataModelOut)
async def create_policy_details(request:Request,
    countryId: str = Body(...),
    policyDetails: List[dict] = Body(...),
    leaveTypeId: str = Body(...),
    branchIds: List[str] = Body(...),
    cityIds: List[str] = Body(...),
    files: List[UploadFile] = File(None)
    ):

当我使用 postman 的 form-data 选项发送请求时，它为 policyDetails 参数显示“0:value is not a valid dict”。我正在发送 [{"name":"name1","department":"d1"}]。它说不是有效的字典，即使我发送有效的字典。谁可以帮我这个事？DataModelOut 类

class DataModelOut(BaseModel):
    message: str = ""
    id: str = ""
    input_data: dict = None
    result: List[dict] = []
    statusCode: int

Answer 1

根据 FastAPI文档，当包含Files或Form参数时，“您不能将Body期望接收的字段声明为JSON”，因为请求将使用application/x-www-form-urlencoded（或者multipart/form-data，如果包含文件）而不是 来对正文进行编码application/json。因此，您不能同时拥有Form（和/或File）数据和JSON数据。这不是 FastAPI 的限制，而是协议的一部分HTTP。也请您看看这个答案。

\n

如果您files: List[UploadFile] = File()从端点中删除了参数，您将看到客户端的 JSON 请求（使用有效的 JSON 负载作为请求正文）将顺利完成，因为端点需要一个application/json-encoded 的请求正文（因为您有使用 type 声明端点中的每个参数Body，而不是multipart/form-data-encoded 请求正文（如果UploadFile/File参数被定义\xe2\x80\x94，无论您是否将其余参数定义为Body字段，都会出现这种情况，相反，它们是预期的form-data；在这种情况下，也可以使用 type，它是直接从\xe2\x80\x94Form继承的类（请参阅此处）。您还可以在http://127.0.0.1:8000/docs上使用OpenAPI/Swagger UI autodocs进行确认。Body

\n

至于声明诸如policyDetails: List[dict] = Body(...)(甚至policyDetails: dict) 之类的参数，它本质上是期望JSON数据，您不能使用Form字段或Body字段与字段一起执行此操作File（在这种情况下，它们将再次被解释为Form字段，如前所述） 。因此，value is not a valid dict当尝试在字段中发送 JSON 数据（即 adict或listof dict）时Form，请求的Content-Type标头实际上设置为multipart/form-data（在您的情况下），甚至是application/x-www-form-urlencoded，如果只form-data包括在内的话。

\n

解决方案

\n

因此，除了 之外files，您的数据可以作为 stringified 发送JSON，并且在服务器端您可以有一个自定义 pydantic 类，它将给定的字符串转换JSON为 Python 字典并根据模型对其进行验证，如本答案中所述。该files参数应与端点中的模型分开定义。下面是一个演示上述方法的工作示例。

\n

工作示例

\n

应用程序.py

\n

from fastapi import FastAPI, File, UploadFile, Body, status\nfrom pydantic import BaseModel\nfrom typing import Optional, List\nimport json\n\napp = FastAPI()\n\nclass DataModelOut(BaseModel):\n    message: str = None\n    id: str = None\n    input_data: dict = None\n    result: List[dict] = []\n    statusCode: int\n \n \nclass DataModelIn(BaseModel):\n    countryId: str\n    policyDetails: List[dict]\n    leaveTypeId: str\n    branchIds: List[str]\n    cityIds: List[str]\n    \n    @classmethod\n    def __get_validators__(cls):\n        yield cls.validate_to_json\n\n    @classmethod\n    def validate_to_json(cls, value):\n        if isinstance(value, str):\n            return cls(**json.loads(value))\n        return value\n    \n\n@app.post(\'/\', response_model=DataModelOut)\ndef create_policy_details(data: DataModelIn = Body(...), files: Optional[List[UploadFile]] = File(None)):\n    print(\'Files received: \', [f.filename for f in files])\n    return {\'input_data\':data, \'statusCode\': status.HTTP_201_CREATED}\n

\n

更新

\n

在 Pydantic V2 中，引入了许多更改 - 请查看迁移指南以获取更多详细信息，以及此答案，关于@validator。上面的工作示例已更新（见下文）以满足这些更改。

\n

应用程序.py

\n

from fastapi import FastAPI, File, UploadFile, Body, status\nfrom pydantic import BaseModel, model_validator\nfrom typing import Optional, List\nimport json\n\napp = FastAPI()\n\n\nclass DataModelOut(BaseModel):\n    message: str = None\n    id: str = None\n    input_data: dict = None\n    result: List[dict] = []\n    statusCode: int\n \n \nclass DataModelIn(BaseModel):\n    countryId: str\n    policyDetails: List[dict]\n    leaveTypeId: str\n    branchIds: List[str]\n    cityIds: List[str]\n\n    @model_validator(mode=\'before\')\n    @classmethod\n    def validate_to_json(cls, value):\n        if isinstance(value, str):\n            return cls(**json.loads(value))\n        return value\n    \n\n@app.post(\'/\', response_model=DataModelOut)\ndef create_policy_details(data: DataModelIn = Body(...), files: List[UploadFile] = File(...)):\n    print(\'Files received: \', [f.filename for f in files])\n    return {\'input_data\': data.model_dump(), \'statusCode\': status.HTTP_201_CREATED}\n

\n

测试上面的例子

\n

也是基于这个答案。

\n

测试.py

\n

import requests\n\nurl = \'http://127.0.0.1:8000/\'\nfiles = [(\'files\', open(\'a.txt\', \'rb\')), (\'files\', open(\'b.txt\', \'rb\'))]\ndata = {\'data\' : \'{"countryId": "US", "policyDetails":  [{"name":"name1","department":"d1"}], "leaveTypeId": "some_id", "branchIds": ["b1", "b2"], "cityIds": ["c1", "c2"]}\'}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n

\n

或者，如果您喜欢这种方式：

\n

import requests\nimport json\n\nurl = \'http://127.0.0.1:8000/\'\nfiles = [(\'files\', open(\'a.txt\', \'rb\')), (\'files\', open(\'b.txt\', \'rb\'))]\ndata_dict = {"countryId": "US", "policyDetails":  [{"name":"name1","department":"d1"}], "leaveTypeId": "some_id", "branchIds": ["b1", "b2"], "cityIds": ["c1", "c2"]}\ndata = {\'data\': json.dumps(data_dict)}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n

\n

您还可以使用http://127.0.0.1:8000/docs上的OpenAPI/Swagger UI 自动文档来测试该应用程序。

\n

最后，您可能会发现这个答案和这个答案很有帮助，其中提供了相关示例。

\n

Answer 2

orm_mode我认为您应该在 Schema/Model 类中添加一个设置为 True 的配置类

    class DataModelOut(BaseModel):
       message: str = ""
       id: str = ""
       input_data: dict = None
       result: List[dict] = []
       statusCode: int
       
       class Config:
         orm_mode = True

Answer 3

问题直接来自 response_model和您的返回值，假设我有一个这样的应用程序

class Example(BaseModel):
    name: str 
    
@app.post("/", response_model=Example)
async def example(value: int):
    return value

现在我正在向此发送请求

pydantic.error_wrappers.ValidationError: 1 validation error for Example
response
  value is not a valid dict (type=type_error.dict)

错误和你的一样。即使我发送相同的参数也会引发相同的错误

class Example(BaseModel):
    name: int 
    other: int

@app.post("/", response_model=Example)
async def example(name: int, other: int):
    return name

Out:   value is not a valid dict (type=type_error.dict)

但是，如果我像这样声明查询参数（文档中的最佳实践），它就会正常工作。

class Example(BaseModel):
    name: int 
    other: int

@app.post("/", response_model=Example)
async def example(ex: Example = Body(...)):
    return ex

Out: {
"name": 0,
"other": 0
}

在您的情况下，您可以创建两个单独的模型，DataModelIn并且DataModelOut，

class DataModelOut(BaseModel):
    message: str = ""
    id: str = ""
    input_data: dict = None
    result: List[dict] = []
    statusCode: int
    
class DataModelIn(BaseModel):
    countryId: str 
    policyDetails: List[dict]
    leaveTypeId: str 
    branchIds: List[str]
    cityIds: List[str]


@app.post("/", response_model=DataModelOut)
async def create_policy_details(data: DataModelIn = Body(...)):
    return {"input_data":data,
            "statusCode":1}

现在我正在向此发送请求

Out: {
  "message": "",
  "id": "",
  "input_data": {
    "countryId": "30",
    "policyDetails": [
      {
        "some": "details"
      }
    ],
    "leaveTypeId": "string",
    "branchIds": [
      "string"
    ],
    "cityIds": [
      "string"
    ]
  },
  "result": [],
  "statusCode": 1
}

它就像一个魅力。您还可以使用response_model_exclude_unset=True参数来丢弃message和id响应，也可以检查一下