问题：

如何在FastAPI POST请求中同时添加文件和JSON主体？

姜献

2023-03-14

具体来说，我希望下面的示例可以工作：

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

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

共有3个答案

越运锋

2023-03-14

我选择了@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)

在endpoint中使用它时，可以使用functools。部分绑定特定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

单勇

2023-03-14

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

每个FastAPI留档：

警告：您可以在路径操作中声明多个File和form参数，但您也不能声明您期望接收为JSON的身体字段，因为请求将使用多部分/form-data而不是应用程序/json对身体进行编码。这不是FastAPI的限制，它是HTTP协议的一部分。

但是，您可以使用Form（…）作为变通方法，将额外的字符串附加为表单数据：

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

尤钱明

2023-03-14

根据 FastAPI 文档：

您可以在路径操作中声明多个表单参数，但也不能将预期接收的 Body 字段声明为 JSON，因为请求将使用 application/x-www-form-urlencoding 而不是 application/json 对正文进行编码（当表单包含文件时，它被编码为多部分/表单数据）。

这不是FastAPI的限制，而是< code>HTTP协议的一部分。

如下所述，可以使用File和form字段同时定义文件和表单数据。下面是一个工作示例。如果您有大量参数，并且想从endpoint单独定义它们，请查看有关如何创建自定义依赖类的答案。

app.py公司

from fastapi import Form, File, UploadFile, Request, FastAPI
from typing import List
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")


@app.post("/submit")
def submit(
    name: str = Form(...),
    point: float = Form(...),
    is_accepted: bool = Form(...),
    files: List[UploadFile] = File(...),
):
    return {
        "JSON Payload ": {"name": name, "point": point, "is_accepted": is_accepted},
        "Filenames": [file.filename for file in files],
    }


@app.get("/", response_class=HTMLResponse)
def main(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

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

模板/index.html

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

也可以使用http://127.0.0.1:8000/docs上的交互式OpenAPI文档(由Swagger UI提供)或者Python请求进行测试，如下所示:

test.py

import requests

url = 'http://127.0.0.1:8000/submit'
files = [('files', open('test_files/a.txt', 'rb')), ('files', open('test_files/b.txt', 'rb'))]
payload ={"name": "foo", "point": 0.13, "is_accepted": False}
resp = requests.post(url=url, data=payload, files=files) 
print(resp.json())

可以使用Pydantic模型以及依赖关系来通知“提交”路径(在下面的情况中),参数化变量< code>base依赖于< code>Base类。请注意，此方法期望< code>base数据作为< code>query(而不是< code>body)参数(然后使用< code >将其转换为等效的< code>JSON有效负载)。dict()方法)和< code >文件作为主体中的< code>multipart/form-data。

app.py公司

from fastapi import Form, File, UploadFile, Request, FastAPI, Depends
from typing import List
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing import Optional
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")


class Base(BaseModel):
    name: str
    point: Optional[float] = None
    is_accepted: Optional[bool] = False


@app.post("/submit")
def submit(base: Base = Depends(), files: List[UploadFile] = File(...)):
    return {
        "JSON Payload ": base.dict(),
        "Filenames": [file.filename for file in files],
    }


@app.get("/", response_class=HTMLResponse)
def main(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

同样，您可以使用下面的模板来测试它，这一次，它使用Javascript来修改< code>form的< code>action属性，以便将< code>form数据作为< code>query参数传递给URL。

模板/index.html

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

如前所述，您还可以使用 Swagger UI 或 Python 请求，如下例所示。注意：这次使用 params=payload，因为参数是查询参数，而不是像前面的方法那样的表单数据（正文）参数。

test.py

import requests

url = 'http://127.0.0.1:8000/submit'
files = [('files', open('test_files/a.txt', 'rb')), ('files', open('test_files/b.txt', 'rb'))]
payload ={"name": "foo", "point": 0.13, "is_accepted": False}
resp = requests.post(url=url, params=payload, files=files)
print(resp.json())

另一种选择是以 JSON 字符串的形式将正文数据作为单个参数（形式类型）传递。在服务器端，您可以创建一个依赖函数，您可以在其中使用 parse_raw 方法解析数据并根据相应的模型验证数据。如果引发验证错误，则会将HTTP_422_UNPROCESSABLE_ENTITY错误（包括错误消息）发送回客户端。示例如下：

app.py公司

from fastapi import FastAPI, status, Form, UploadFile, File, Depends, Request
from pydantic import BaseModel, ValidationError
from fastapi.exceptions import HTTPException
from fastapi.encoders import jsonable_encoder
from typing import Optional, List
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse

app = FastAPI()
templates = Jinja2Templates(directory="templates")


class Base(BaseModel):
    name: str
    point: Optional[float] = None
    is_accepted: Optional[bool] = False


def checker(data: str = Form(...)):
    try:
        model = Base.parse_raw(data)
    except ValidationError as e:
        raise HTTPException(
            detail=jsonable_encoder(e.errors()),
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        )

    return model


@app.post("/submit")
def submit(model: Base = Depends(checker), files: List[UploadFile] = File(...)):
    return {"JSON Payload ": model, "Filenames": [file.filename for file in files]}


@app.get("/", response_class=HTMLResponse)
def main(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

如果您有多个模型，并且希望避免为每个模型创建一个checker函数，那么您可以创建一个checker类，如文档中所述，并且拥有一个模型字典，您可以使用它来查找要解析的模型。示例:

# ...
models = {"base": Base, "other": SomeOtherModel}

class DataChecker:
    def __init__(self, name: str):
        self.name = name

    def __call__(self, data: str = Form(...)):
        try:
            model = models[self.name].parse_raw(data)
        except ValidationError as e:
            raise HTTPException(
                detail=jsonable_encoder(e.errors()),
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            )
        return model

checker = DataChecker("base")
checker2 = DataChecker("other")

@app.post("/submit")
def submit(model: Base = Depends(checker), files: List[UploadFile] = File(...)):
    # ...

test.py

请注意，在< code>JSON中，布尔值使用小写的< code>true或< code>false文字表示，而在Python中，它们必须大写为< code>True或< code>False。

import requests

url = 'http://127.0.0.1:8000/submit'
files = [('files', open('test_files/a.txt', 'rb')), ('files', open('test_files/b.txt', 'rb'))]
data = {'data': '{"name": "foo", "point": 0.13, "is_accepted": false}'}
resp = requests.post(url=url, data=data, files=files) 
print(resp.json())

或者，如果您愿意：

import requests
import json

url = 'http://127.0.0.1:8000/submit'
files = [('files', open('test_files/a.txt', 'rb')), ('files', open('test_files/b.txt', 'rb'))]
data = {'data': json.dumps({"name": "foo", "point": 0.13, "is_accepted": False})}
resp = requests.post(url=url, data=data, files=files) 
print(resp.json())

使用 Fetch API 或 Axios 进行测试

模板/index.html

<!DOCTYPE html>
<html>
   <head>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/axios/0.27.2/axios.min.js"></script>
   </head>
   <body>
      <input type="file" id="fileInput" name="file" multiple><br>
      <input type="button" value="Submit using fetch" onclick="submitUsingFetch()">
      <input type="button" value="Submit using axios" onclick="submitUsingAxios()">
      <script>
         function submitUsingFetch() {
             var fileInput = document.getElementById('fileInput');
             if (fileInput.files[0]) {
                var formData = new FormData();
                formData.append("data", JSON.stringify({"name": "foo", "point": 0.13, "is_accepted": false}));
                for (const file of fileInput.files)
                    formData.append('files', file);
                    
                 fetch('/submit', {
                       method: 'POST',
                       body: formData,
                     })
                     .then(response => {
                       console.log(response);
                     })
                     .catch(error => {
                       console.error(error);
                     });
             }
         }
         
         function submitUsingAxios() {
             var fileInput = document.getElementById('fileInput');
             if (fileInput.files[0]) {
                var formData = new FormData();
                formData.append("data", JSON.stringify({"name": "foo", "point": 0.13, "is_accepted": false}));
                for (const file of fileInput.files)
                    formData.append('files', file);
                    
                 axios({
                         method: 'POST',
                         url: '/submit',
                         data: formData,
                     })
                     .then(response => {
                       console.log(response);
                     })
                     .catch(error => {
                       console.error(error);
                     });
             }
         }
      </script>
   </body>
</html>

另一个方法来自这里的讨论，它将一个自定义类与一个classmethod结合在一起，用于将给定的＜code＞JSON＜/code＞字符串转换为Python字典，然后使用该字典对Pydantic模型进行验证。与上面的方法3类似，输入数据应该以＜code＞JSON＜/code＞字符串的形式作为单个＜code＞Form＜/code＞参数传递（用＜code＞Body＜/code>类型定义参数也可以，并且JSON字符串仍然需要＜code＞Form＜/code＞数据，因为在这种情况下，数据编码为＜code＞multipart/Form data＜/ccode＞）。因此，同样的测试。py文件和索引。前面方法中的html模板可以用于测试下面的。

app.py公司

from fastapi import FastAPI, File, Body, UploadFile, Request
from pydantic import BaseModel
from typing import Optional, List
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
import json

app = FastAPI()
templates = Jinja2Templates(directory="templates")


class Base(BaseModel):
    name: str
    point: Optional[float] = None
    is_accepted: Optional[bool] = False

    @classmethod
    def __get_validators__(cls):
        yield cls.validate_to_json

    @classmethod
    def validate_to_json(cls, value):
        if isinstance(value, str):
            return cls(**json.loads(value))
        return value


@app.post("/submit")
def submit(data: Base = Body(...), files: List[UploadFile] = File(...)):
    return {"JSON Payload ": data, "Filenames": [file.filename for file in files]}


@app.get("/", response_class=HTMLResponse)
def main(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

如何在FastAPI POST请求中同时添加文件和JSON主体？

共有3个答案

相关问答

相关文章

相关阅读

相关工具

相关文档