如何在 FastAPI POST 请求中同时添加文件和 JSON 正文?

ASa*_*mil 2 python http http-post pydantic fastapi

具体来说,我希望以下示例能够正常工作:

from typing import List
from pydantic import BaseModel
from fastapi import FastAPI, UploadFile, File


app = FastAPI()


class DataConfiguration(BaseModel):
    textColumnNames: List[str]
    idColumn: str


@app.post("/data")
async def data(dataConfiguration: DataConfiguration,
               csvFile: UploadFile = File(...)):
    pass
    # read requested id and text columns from csvFile
Run Code Online (Sandbox Code Playgroud)

如果这不是 POST 请求的正确方法,请告诉我如何从 FastAPI 中上传的 CSV 文件中选择所需的列。

Chr*_*ris 93

根据FastAPI 文档

\n
\n

您可以Form在路径操作中声明多个参数,但您\n不能也将Body您期望接收的字段JSON声明为,因为\n请求的正文将使用\n 编码,application/x-www-form-urlencoded而不是application/json(当表单包含文件时,它是编码为multipart/form-data)。

\n

这不是 FastAPI 的限制,而是协议的一部分HTTP

\n
\n

python-multipart请注意,如果您尚未安装\xe2\x80\x94,则需要先安装\xe2\x80\x94 ,因为上传的文件是作为“表单数据”发送的。例如:

\n
pip install python-multipart\n
Run Code Online (Sandbox Code Playgroud)\n

还应该注意的是,在下面的示例中,端点是用 normal 定义的def,但您也可以使用async def(根据您的需要)。请查看此答案,def了解有关 FastAPI 中vs的更多详细信息async def

\n

如果您正在寻找如何上传文件和list字典/JSON数据,请查看这个答案,以及这个答案这个工作示例的答案(主要基于以下一些方法)。

\n

方法一

\n

如此处所述,可以使用File和同时定义文件和表单字段Form。下面是一个工作示例。如果您有大量参数并且希望与端点分开定义它们,请查看有关如何创建自定义依赖项类的答案。

\n

应用程序.py

\n
pip install python-multipart\n
Run Code Online (Sandbox Code Playgroud)\n

您可以通过访问下面的模板来测试上面的示例http://127.0.0.1:8000。如果您的模板不包含任何 Jinja 代码,您也可以返回一个简单的HTMLResponse.

\n

模板/index.html

\n
from fastapi import Form, File, UploadFile, Request, FastAPI\nfrom typing import List\nfrom fastapi.responses import HTMLResponse\nfrom fastapi.templating import Jinja2Templates\n\napp = FastAPI()\ntemplates = Jinja2Templates(directory="templates")\n\n\n@app.post("/submit")\ndef submit(\n    name: str = Form(...),\n    point: float = Form(...),\n    is_accepted: bool = Form(...),\n    files: List[UploadFile] = File(...),\n):\n    return {\n        "JSON Payload": {"name": name, "point": point, "is_accepted": is_accepted},\n        "Filenames": [file.filename for file in files],\n    }\n\n\n@app.get("/", response_class=HTMLResponse)\ndef main(request: Request):\n    return templates.TemplateResponse("index.html", {"request": request})\n
Run Code Online (Sandbox Code Playgroud)\n

您还可以使用交互式OpenAPI/Swagger UI 自动文档/docs(例如 )http://127.0.0.1:8000/docs或使用 Python来测试此示例requests,如下所示:

\n

测试.py

\n
<!DOCTYPE html>\n<html>\n   <body>\n      <form method="post" action="http://127.0.0.1:8000/submit"  enctype="multipart/form-data">\n         name : <input type="text" name="name" value="foo"><br>\n         point : <input type="text" name="point" value=0.134><br>\n         is_accepted : <input type="text" name="is_accepted" value=True><br>    \n         <label for="file">Choose files to upload</label>\n         <input type="file" id="files" name="files" multiple>\n         <input type="submit" value="submit">\n      </form>\n   </body>\n</html>\n
Run Code Online (Sandbox Code Playgroud)\n

方法二

\n

人们还可以使用 Pydantic 模型以及Dependency来通知/submit端点(在下面的示例中)参数化变量base取决于Base类。请注意,此方法期望base数据作为query而不是 body)参数,然后对这些数据进行验证并将其转换为 Pydantic 模型(在本例中,即模型Base)。从 FastAPI 端点返回 Pydantic 模型实例(在本例中为base)将使用 ,在幕后自动转换为等效的字典/JSON 对象,如本答案jsonable_encoder中详细解释的。但是,如果您希望自己在端点内完成此操作,则可以使用 Pydantic 的方法,例如,或简单地,如此答案中所述。此外,以下示例期望请求正文中的内容相同。model_dump()base.model_dump()dict(base)dataFilesmultipart/form-data

\n

应用程序.py

\n
import requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {"name": "foo", "point": 0.13, "is_accepted": False}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n

同样,您可以使用下面的模板对其进行测试,这一次,该模板使用 JavaScript 修改元素action的属性form,以便将form数据作为query参数传递到 URL 而不是form-data.

\n

模板/index.html

\n
from fastapi import Form, File, UploadFile, Request, FastAPI, Depends\nfrom typing import List\nfrom fastapi.responses import HTMLResponse\nfrom pydantic import BaseModel\nfrom typing import Optional\nfrom fastapi.templating import Jinja2Templates\n\napp = FastAPI()\ntemplates = Jinja2Templates(directory="templates")\n\n\nclass Base(BaseModel):\n    name: str\n    point: Optional[float] = None\n    is_accepted: Optional[bool] = False\n\n\n@app.post("/submit")\ndef submit(base: Base = Depends(), files: List[UploadFile] = File(...)):\n    return {\n        "JSON Payload": base,\n        "Filenames": [file.filename for file in files],\n    }\n\n\n@app.get("/", response_class=HTMLResponse)\ndef main(request: Request):\n    return templates.TemplateResponse("index.html", {"request": request})\n
Run Code Online (Sandbox Code Playgroud)\n

如前所述,要测试 API,您还可以使用 Swagger UI 或 Python requests,如下面的示例所示。请注意,数据现在应该传递给方法的params而不是data参数requests.post(),因为数据现在作为query参数发送,而不是在请求正文中发送,这是前面方法 1form-data中的情况。

\n

测试.py

\n
<!DOCTYPE html>\n<html>\n   <body>\n      <form method="post" id="myForm" onclick="transformFormData();" enctype="multipart/form-data">\n         name : <input type="text" name="name" value="foo"><br>\n         point : <input type="text" name="point" value=0.134><br>\n         is_accepted : <input type="text" name="is_accepted" value=True><br>    \n         <label for="file">Choose files to upload</label>\n         <input type="file" id="files" name="files" multiple>\n         <input type="submit" value="submit">\n      </form>\n      <script>\n         function transformFormData(){\n            var myForm = document.getElementById(\'myForm\');\n            var qs = new URLSearchParams(new FormData(myForm)).toString();\n            myForm.action = \'http://127.0.0.1:8000/submit?\' + qs;\n         }\n      </script>\n   </body>\n</html>\n
Run Code Online (Sandbox Code Playgroud)\n

方法三

\n

另一种选择是将正文数据作为FormJSON 字符串形式的单个参数(类型为 )传递。为此,您需要在服务器端创建依赖函数。

\n

依赖项是“只是一个函数,它可以采用路径操作函数(也称为端点)可以采用的所有相同参数。您可以将其视为没有装饰器的路径操作函数” 。因此,您需要以与端点参数相同的方式声明依赖项(即,依赖项中的参数名称和类型应该是客户端向该端点发送 HTTP 请求时 FastAPI 所期望的名称和类型) , 例如,data: str = Form(...))。然后,base在端点中创建一个新参数(例如,),使用Depends()依赖函数并将其作为参数传递给它(注意:不要直接调用它,这意味着不要在函数末尾添加括号\ 的名称,而是使用例如 ,Depends(checker)其中checker是依赖函数的名称)。每当新请求到达时,FastAPI 将负责调用您的依赖项、获取结果并将该结果分配给端点base中的参数(例如 )。有关依赖项的更多详细信息,请查看本节中提供的链接。

\n

在这种情况下,应该使用依赖函数来使用data方法parse_raw注意:在 Pydantic V2 中parse_raw已弃用并替换为model_validate_json)来解析(JSON 字符串),并data根据相应的 Pydantic 模型进行验证。如果ValidationError引发,HTTP_422_UNPROCESSABLE_ENTITY则应将错误发送回客户端,包括错误消息;否则,该模型的实例(即Base本例中的模型)将被分配给端点中的参数,可以根据需要使用该参数。示例如下:

\n

应用程序.py

\n
import requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\nparams = {"name": "foo", "point": 0.13, "is_accepted": False}\nresp = requests.post(url=url, params=params, files=files)\nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n
通用Checker依赖类
\n

如果您有多个模型并且希望避免checker为每个模型创建函数,您可以创建一个通用 Checker依赖项类,如文档中所述(也请参阅此答案以获取更多详细信息),并将其用于每个不同的模型在你的 API 中。例子:

\n
from fastapi import FastAPI, status, Form, UploadFile, File, Depends, Request\nfrom pydantic import BaseModel, ValidationError\nfrom fastapi.exceptions import HTTPException\nfrom fastapi.encoders import jsonable_encoder\nfrom typing import Optional, List\nfrom fastapi.templating import Jinja2Templates\nfrom fastapi.responses import HTMLResponse\n\napp = FastAPI()\ntemplates = Jinja2Templates(directory="templates")\n\n\nclass Base(BaseModel):\n    name: str\n    point: Optional[float] = None\n    is_accepted: Optional[bool] = False\n\n\ndef checker(data: str = Form(...)):\n    try:\n        return Base.model_validate_json(data)\n    except ValidationError as e:\n        raise HTTPException(\n            detail=jsonable_encoder(e.errors()),\n            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,\n        )\n\n\n@app.post("/submit")\ndef submit(base: Base = Depends(checker), files: List[UploadFile] = File(...)):\n    return {"JSON Payload": base, "Filenames": [file.filename for file in files]}\n\n\n@app.get("/", response_class=HTMLResponse)\ndef main(request: Request):\n    return templates.TemplateResponse("index.html", {"request": request})\n
Run Code Online (Sandbox Code Playgroud)\n
任意 JSON 数据
\n

如果针对特定 Pydantic 模型验证输入数据对您来说并不重要,但是您希望接收任意JSON 数据并简单地检查客户端是否发送了有效的 JSON 字符串,您可以使用以下:

\n
# ...  rest of the code is the same as above\n\nclass Other(BaseModel):\n    msg: str\n    details: Base\n \n    \nclass Checker:\n    def __init__(self, model: BaseModel):\n        self.model = model\n\n    def __call__(self, data: str = Form(...)):\n        try:\n            return self.model.model_validate_json(data)\n        except ValidationError as e:\n            raise HTTPException(\n                detail=jsonable_encoder(e.errors()),\n                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,\n            )\n\n\n@app.post("/submit")\ndef submit(base: Base = Depends(Checker(Base)), files: List[UploadFile] = File(...)):\n    pass\n\n\n@app.post("/submit_other")\ndef submit_other(other: Other = Depends(Checker(Other)), files: List[UploadFile] = File(...)):\n    pass\n
Run Code Online (Sandbox Code Playgroud)\n

或者,您可以简单地使用JsonPydantic 中的类型(如此所示):

\n
# ...\nfrom json import JSONDecodeError\nimport json\n\ndef checker(data: str = Form(...)):\n    try:\n       return json.loads(data)\n    except JSONDecodeError:\n        raise HTTPException(status_code=400, detail=\'Invalid JSON data\')\n\n\n@app.post("/submit")\ndef submit(payload: dict = Depends(checker), files: List[UploadFile] = File(...)):\n    pass\n
Run Code Online (Sandbox Code Playgroud)\n

使用Python进行测试requests

\n

测试.py

\n

请注意,在 中JSON,布尔值使用小写的truefalse文字表示,而在 Python 中,它们必须大写为TrueFalse

\n
import requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {\'data\': \'{"name": "foo", "point": 0.13, "is_accepted": false}\'}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n

或者,如果您愿意:

\n
from pydantic import Json\n\n@app.post("/submit")\ndef submit(data: Json = Form(), files: List[UploadFile] = File(...)):\n    pass\n
Run Code Online (Sandbox Code Playgroud)\n

PS 要使用 Python测试/submit_other端点(在前面的泛型类中描述) ,请将上例中的属性替换为以下属性:Checkerrequestsdata

\n
import requests\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {\'data\': \'{"name": "foo", "point": 0.13, "is_accepted": false}\'}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n

使用 Fetch API 或 Axios 进行测试

\n

模板/index.html

\n
import requests\nimport json\n\nurl = \'http://127.0.0.1:8000/submit\'\nfiles = [(\'files\', open(\'test_files/a.txt\', \'rb\')), (\'files\', open(\'test_files/b.txt\', \'rb\'))]\ndata = {\'data\': json.dumps({"name": "foo", "point": 0.13, "is_accepted": False})}\nresp = requests.post(url=url, data=data, files=files) \nprint(resp.json())\n
Run Code Online (Sandbox Code Playgroud)\n

方法4

\n

另一种方法来自github 讨论,并将自定义类与用于将给定字符串转换为 Python 字典的类方法结合JSON在一起,然后将其用于针对 Pydantic 模型进行验证(请注意,与前面提到的 github 链接,下面的例子使用的是@model_validator(mode=\'before\'),因为引入了 Pydantic V2)。

\n

与上面的方法 3类似,输入数据应作为字符串Form形式的单个参数传递JSON(请注意,data在下面的示例中使用Bodyor定义参数Form都可以工作,无论 \xe2\x80\x94Form是直接继承自的类Body也就是说,FastAPI 仍然期望 JSON 字符串作为form数据,而不是application/json,因为在这种情况下,请求将使用multipart/form-data) 编码正文。因此,上面方法 3中的相同test.py示例和index.html模板也可以用于测试下面的示例。

\n

应用程序.py

\n
import requests\nimport json\n\nurl = \'http://127.0.0.1:8000/submit_other\'\ndata = {\'data\': json.dumps({"msg": "Hi", "details": {"name": "bar", "point": 0.11, "is_accepted": True}})}\n# ... rest of the code is the same as above\n
Run Code Online (Sandbox Code Playgroud)\n

方法5

\n

另一种解决方案是将文件字节转换为base64-format 字符串,并将其与您可能想要发送到服务器的其他数据一起添加到 JSON 对象中。我不会强烈推荐使用这种方法;但是,为了完整起见,它已作为进一步的选项添加到此答案中。

\n

我不建议使用它的原因是,使用编码文件base64本质上会增加文件的大小,从而增加带宽利用率,以及上传文件所需的时间和资源(例如,CPU 使用率) (特别是当多个用户同时使用 API 时),因为需要分别在客户端和服务器端进行 Base64 编码和解码(这种方法仅适用


Eri*_*c L 5

您不能将表单数据与 json 混合使用。

根据 FastAPI文档

警告:您可以声明多个FileForm参数的路径运行,但你不能同时申报Body,您希望收到的JSON字段,请求将身体用编码multipart/form-data代替application/json。这不是 FastAPI 的限制,它是 HTTP 协议的一部分。

但是,您可以Form(...)将额外的字符串附加为form-data

from typing import List
from fastapi import FastAPI, UploadFile, File, Form


app = FastAPI()


@app.post("/data")
async def data(textColumnNames: List[str] = Form(...),
               idColumn: str = Form(...),
               csvFile: UploadFile = File(...)):
    pass
Run Code Online (Sandbox Code Playgroud)


phi*_*rse 5

我采用了@Chris 的非常优雅的Method3(最初由@M.Winkwns 提出)。不过,我稍微修改了它以适用于任何Pydantic 模型:

from typing import Type, TypeVar

from pydantic import BaseModel, ValidationError
from fastapi import Form

Serialized = TypeVar("Serialized", bound=BaseModel)


def form_json_deserializer(schema: Type[Serialized], data: str = Form(...)) -> Serialized:
    """
    Helper to serialize request data not automatically included in an application/json body but
    within somewhere else like a form parameter. This makes an assumption that the form parameter with JSON data is called 'data'

    :param schema: Pydantic model to serialize into
    :param data: raw str data representing the Pydantic model
    :raises ValidationError: if there are errors parsing the given 'data' into the given 'schema'
    """
    try:
        return schema.parse_raw(data)
    except ValidationError as e 
        raise HTTPException(detail=jsonable_encoder(e.errors()), status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)

    
Run Code Online (Sandbox Code Playgroud)

当您在端点中使用它时,您可以用来functools.partial绑定特定的 Pydantic 模型:

import functools

from pydantic import BaseModel
from fastapi import Form, File, UploadFile, FastAPI

class OtherStuff(BaseModel):
    stuff: str

class Base(BaseModel):
    name: str
    stuff: OtherStuff

@app.post("/upload")
async def upload(
    data: Base = Depends(functools.partial(form_json_deserializer, Base)),
    files: Sequence[UploadFile] = File(...)
) -> Base:
    return data
Run Code Online (Sandbox Code Playgroud)


归档时间:

查看次数:

1659 次

最近记录:

4 年,4 月 前