具体来说，我希望下面的示例能够正常工作：
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 请求方式，请告诉我如何从上传的 CSV 文件中选择所需的列在 FastAPI 中。pip install python-multipart
需要注意的是，在下面的示例中，端点是使用普通的def定义的，但您也可以根据需要使用async def。有关def与async def在FastAPI中的更多详细信息，请参阅this answer。
如此处所述，可以同时使用File和Form定义文件和表单字段。以下是一个工作示例。如果您有大量参数，并且希望将它们与端点分开定义，请参阅this answer，了解如何创建自定义依赖类。
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代码，你也可以返回一个简单的HTMLResponse。
templates/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>
/docs，例如：http://127.0.0.1:8000/docs，或者使用Python的requests库，如下所示：
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'))]
data = {"name": "foo", "point": 0.13, "is_accepted": False}
resp = requests.post(url=url, data=data, files=files)
print(resp.json())
/submit端点（在下面的示例中），参数化变量base依赖于Base类。请注意，此方法期望将base数据作为query（而不是body）参数，然后对其进行验证并转换为Pydantic模型（在本例中即Base模型）。从FastAPI端点返回一个Pydantic模型实例（在本例中即base）将自动转换为等效的字典/JSON对象，使用jsonable_encoder在后台进行处理，详细说明请参见this answer。然而，如果您希望在端点内部自行完成此操作，可以使用Pydantic的model_dump()方法，例如base.model_dump()，或者简单地使用dict(base)，详细说明请参见this answer。除了data之外，以下示例还期望将Files作为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,
"Filenames": [file.filename for file in files],
}
@app.get("/", response_class=HTMLResponse)
def main(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
form元素的action属性，以便将form数据作为query参数传递到URL，而不是form-data。
templates/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>
params（而不是data）参数，这是requests.post()方法的参数，因为数据现在作为query参数发送，而不是在请求体中作为form-data，这是之前的方法1的情况。
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'))]
params = {"name": "foo", "point": 0.13, "is_accepted": False}
resp = requests.post(url=url, params=params, files=files)
print(resp.json())
另一个选择是将请求体数据作为一个参数（类型为Form）以JSON字符串的形式传递。为此，您需要在服务器端创建一个dependency函数。
data，使用parse_raw方法（注意：在Pydantic V2中，parse_raw已被弃用，并被model_validate_json替代），同时验证data与相应的Pydantic模型。如果引发了ValidationError，则应将HTTP_422_UNPROCESSABLE_ENTITY错误发送回客户端，包括错误消息；否则，将该模型的实例（即，在这种情况下是Base模型）分配给端点中的参数，可以根据需要使用。以下是示例：
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:
return Base.model_validate_json(data)
except ValidationError as e:
raise HTTPException(
detail=jsonable_encoder(e.errors()),
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
)
@app.post("/submit")
def submit(base: Base = Depends(checker), files: List[UploadFile] = File(...)):
return {"JSON Payload": base, "Filenames": [file.filename for file in files]}
@app.get("/", response_class=HTMLResponse)
def main(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
# ...
models = {"base": Base, "other": SomeOtherModel}
class DataChecker:
def __init__(self, name: str):
self.name = name
def __call__(self, data: str = Form(...)):
try:
return models[self.name].model_validate_json(data)
except ValidationError as e:
raise HTTPException(
detail=jsonable_encoder(e.errors()),
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
)
base_checker = DataChecker("base")
other_checker = DataChecker("other")
@app.post("/submit")
def submit(base: Base = Depends(base_checker), files: List[UploadFile] = File(...)):
pass
# ...
import json
def checker(data: str = Form(...)):
try:
return json.loads(data)
except JSONDecodeError:
raise HTTPException(status_code=400, detail='Invalid JSON data')
@app.post("/submit")
def submit(d: dict = Depends(checker), files: List[UploadFile] = File(...)):
pass
test.py
请注意，在 JSON 中，布尔值使用小写的 true 或 false 字面量表示，而在 Python 中，它们必须大写，即 True 或 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())
templates/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" onchange="reset()" multiple><br>
<input type="button" value="Submit using fetch" onclick="submitUsingFetch()">
<input type="button" value="Submit using axios" onclick="submitUsingAxios()">
<p id="resp"></p>
<script>
function reset() {
var resp = document.getElementById("resp");
resp.innerHTML = "";
resp.style.color = "black";
}
function submitUsingFetch() {
var resp = document.getElementById("resp");
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 => response.json())
.then(data => {
resp.innerHTML = JSON.stringify(data); // data is a JSON object
})
.catch(error => {
console.error(error);
});
} else {
resp.innerHTML = "Please choose some file(s)...";
resp.style.color = "red";
}
}
function submitUsingAxios() {
var resp = document.getElementById("resp");
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 => {
resp.innerHTML = JSON.stringify(response.data); // response.data is a JSON object
})
.catch(error => {
console.error(error);
});
} else {
resp.innerHTML = "Please choose some file(s)...";
resp.style.color = "red";
}
}
</script>
</body>
</html>
另一种方法来自于github的讨论这里，它使用了一个自定义类custom class，其中包含一个classmethod，用于将给定的JSON字符串转换为Python字典，然后用于对Pydantic模型进行验证（注意，与上述github链接中给出的示例相比，下面的示例使用了@model_validator(mode='before')，因为Pydantic V2的引入）。
Form是直接继承自Body的类}}。也就是说，FastAPI仍然期望将JSON字符串作为“form”数据，而不是“application/json”，在这种情况下，请求的主体将使用“multipart/form-data”进行编码）。因此，与上面的“方法3”一样，可以使用相同的“test.py”示例和“index.html”模板来测试下面的示例。
app.py
from fastapi import FastAPI, File, Body, UploadFile, Request
from pydantic import BaseModel, model_validator
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
@model_validator(mode='before')
@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})
另一种解决方案是将文件字节转换为base64格式的字符串，并将其添加到JSON对象中，以及您可能想要发送到服务器的其他数据。我不强烈推荐使用这种方法；然而，出于完整性的考虑，它已被添加到这个答案中作为另一种选择。
我不建议使用它的原因是使用base64编码文件会增加文件的大小，从而增加带宽利用率，以及上传文件所需的时间和资源（例如，CPU使用率）（特别是当API将被多个用户同时使用时），因为base64编码和解码将分别在客户端和服务器端进行（这种方法只对非常小的图像有用）。根据MDN的文档：
使用这种方法，再次强烈不建议，原因如上所述，您需要确保使用普通的def来定义端点，因为base64.b64decode()执行的是一个阻塞操作，会阻塞事件循环，从而阻塞整个服务器。请参考this answer了解更多细节。否则，要使用async def端点，您应该在外部的ThreadPool或ProcessPool中执行解码函数（同样，请参考this answer了解如何执行），并使用aiofiles将文件写入磁盘（还请参考this answer）。
下面的示例提供了Python requests和JavaScript的客户端测试示例。
app.py
from fastapi import FastAPI, Request, HTTPException
from typing import List
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from fastapi.templating import Jinja2Templates
import base64
import binascii
app = FastAPI()
templates = Jinja2Templates(directory='templates')
class Bas64File(BaseModel):
filename: str
owner: str
bas64_str: str
@app.post('/submit')
def submit(files: List[Bas64File]):
for file in files:
try:
contents = base64.b64decode(file.bas64_str.encode('utf-8'))
with open(file.filename, 'wb') as f:
f.write(contents)
except base64.binascii.Error as e:
raise HTTPException(
400, detail='There was an error decoding the base64 string'
)
except Exception:
raise HTTPException(
500, detail='There was an error uploading the file(s)'
)
return {'Filenames': [file.filename for file in files]}
@app.get('/', response_class=HTMLResponse)
def main(request: Request):
return templates.TemplateResponse('index.html', {'request': request})
test.py
import requests
import os
import glob
import base64
url = 'http://127.0.0.1:8000/submit'
paths = glob.glob('files/*', recursive=True)
payload = []
for p in paths:
with open(p, 'rb') as f:
bas64_str = base64.b64encode(f.read()).decode('utf-8')
payload.append({'filename': os.path.basename(p), 'owner': 'me', 'bas64_str': bas64_str})
resp = requests.post(url=url, json=payload)
print(resp.json())
templates/index.html
<input type="file" id="fileInput" onchange="base64Handler()" multiple><br>
<script>
async function base64Handler() {
var fileInput = document.getElementById('fileInput');
var payload = [];
for (const file of fileInput.files) {
var dict = {};
dict.filename = file.name;
dict.owner = 'me';
base64String = await this.toBase64(file);
dict.bas64_str = base64String.replace("data:", "").replace(/^.+,/, "");
payload.push(dict);
}
uploadFiles(payload);
}
function toBase64(file) {
return new Promise((resolve, reject) => {
const reader = new FileReader();
reader.readAsDataURL(file);
reader.onload = () => resolve(reader.result);
reader.onerror = error => reject(error);
});
};
function uploadFiles(payload) {
fetch('/submit', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify(payload)
})
.then(response => {
console.log(response);
})
.catch(error => {
console.error(error);
});
}
</script>
@classmethod
def __get_validators__(cls):
yield cls._validate_from_json_string
@classmethod
def _validate_from_json_string(cls, value):
if isinstance(value, str):
return cls.validate(json.loads(value.encode()))
return cls.validate(value) - derphelix你不能将form-data和json混合使用。
根据FastAPI的文档：
警告： 您可以在路径操作中声明多个
File和Form参数，但您不能同时声明期望接收为JSON的Body字段，因为请求将使用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
我采用了 @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)
当你在端点中使用时，你可以利用 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
根据 @Chris 的说法（仅供完整性考虑）：
根据 FastAPI 文档，
您可以在路径操作中声明多个表单参数，但不能同时声明希望以 JSON 格式接收的 Body 字段，因为请求将使用 application/x-www-form-urlencoded 而不是 application/json 进行编码。 （但当表单包含文件时，它会被编码为 multipart/form-data）
这不是 FastAPI 的限制，而是 HTTP 协议的一部分。
由于他的 Method1 不可行，而 Method2 无法处理深层嵌套的数据类型，所以我提出了一个不同的解决方案：
简单地将您的数据类型转换为字符串/JSON，并调用 pydantic 的 parse_raw 函数。
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("/submit")
async def submit(base: str = Form(...), files: List[UploadFile] = File(...)):
try:
model = Base.parse_raw(base)
except pydantic.ValidationError as e:
raise HTTPException(
detail=jsonable_encoder(e.errors()),
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY
) from e
return {"JSON Payload ": received_data, "Uploaded Filenames": [file.filename for file in files]}
使用Pydantic模型来获得更清晰的文档示例。该文件已编码为Base64，也可以应用其他逻辑。
class BaseTestUser(BaseModel):
name: str
image_1920: str
class UpdateUserEncodeFile(BaseTestUser):
def __init__(self, name: str = Form(...), image_1920: UploadFile = File(...)):
super().__init__(name=name, image_1920=base64.b64encode(image_1920.file.read()))
#路由器
@router.put("/users/{id}/encoded", status_code=status.HTTP_200_OK)
def user_update_encode(id: int, user:UpdateUserEncodeFile=Depends()):
return user
如果您正在使用pydantic v2：
import json
@app.post(/endpoint)
async def endpoint(file: UploadFile, payload: A)
class A(BaseModel):
attr: str
@model_validator(mode="before")
@classmethod
def to_py_dict(cls, data):
return json.loads(data)
def checker(data: str = Form(...)):
...
@app.post("/submit")
def submit(data: dict = Depends(checker), files: List[UploadFile] = File(...)):
pass
files = [('files', open('test_files/a.txt', 'rb')), ('files', ...
data = {'data': '{"name":...}
resp = requests.post(url=url, data=data, files=files)
data = {'data': '{"name":...}
checker是一个FastAPI的依赖函数，所以它的参数被用来替代路径操作函数中的参数。（这就是依赖的全部意义：重用参数集合而不是重复定义它们）。它具有与所有路径操作函数相同的形状和结构。
（请注意，你可以把它看作是一个没有"装饰器"（没有@app.get("/some-path")）的路径操作函数。
def submit...是一个"路径操作函数"的示例）data
resp = requests.post(url=url, data=data, files=files)
需要通过请求的post方法来进行（所以如果你改变了这个方法，你很快就会发现）。
同样，files的唯一值必须与在客户端创建的字典中的值以及函数中的参数名匹配。其余的要么是请求post函数的必需参数，要么是任意选择。
别误会 - 用与分配给它的参数相同的名称来调用任意变量，非常符合Python的风格 - 只是这让我在理解Chris的回答时有些困惑。
为了更清楚，我在下面的摘录中替换了“data”这个词。（并添加了一个断言和一个写入...）
服务：
def checker(foo: str = Form(...)):
return "dingo"
...
@app.post("/submit")
def submit(quux: dict = Depends(checker), bananas: List[UploadFile] = File(...)):
assert quux == "dingo" # quux assigned to return value of checker
# write bananas to local files:
在客户端中：
apples = [('bananas', open('test_files/a.txt', 'rb')), ('bananas', ...
baz = {'foo': { 'name': '...'} ... }
resp = requests.post(url=url, data=baz, files=apples)
依赖函数参数checker(foo必须以表单数据{'foo': {...的形式提供。
客户端必须以请求体的字典/JSON中的表单字段名称作为键提供。仔细看看我的代码中出现的两个“foo”和四个“bananas”。
DataConfiguration类链接。 - Gino Mempin