首先，根据 FastAPI 文档 ，您需要安装 python-multipart —如果尚未安装 —因为上传的文件将作为“表单数据”发送。例如：
pip install python-multipart
以下示例使用 .file 的属性 UploadFile 来获取实际的 Python 文件（即 SpooledTemporaryFile ），这允许您调用 SpooledTemporaryFile 的方法，例如 .read() 和 .close() ，而无需 await 它们。但是，在这种情况下，使用 定义端点非常重要 def - 否则，如果端点是用 定义的，则此类操作将阻塞服务器，直到它们完成 async def 。在 FastAPI 中，正常 def 端点 在外部线程池中运行，然后等待该线程池，而不是直接调用（因为它会阻塞服务器） .
FastAPI/Starlette 使用的 SpooledTemporaryFile 属性 max_size 设置为 1 MB，这意味着数据将在内存中缓冲，直到文件大小超过 1 MB，此时数据将写入磁盘上的临时文件（位于操作系统的临时目录下）。因此，如果您上传的文件大于 1 MB，它将不会存储在内存中，并且调用 file.file.read() 实际上会将数据从磁盘读入内存。因此，如果文件太大而无法放入服务器的 RAM 中，您应该分块读取文件并一次处理一个块，如下面 \'分块读取文件\' 部分所述。您也可以看看 这个答案 分块上传大文件 的方法 ，使用 Starlette 的 .stream() 方法和 streaming-form-data 允许解析流式块的包 multipart/form-data ，从而大大缩短上传文件所需的时间。
如果您必须使用 来定义端点 async def — 您可能需要 await 为路由中的其他协程这样做 — 那么您应该使用 异步 读取和写入内容，如此 答案 。此外，如果您需要在上传文件的同时发送其他数据（例如 JSON 数据），请查看 此答案 。我还建议您看看 这个答案 和 def 之间的区别 async def 。
上传单个文件
应用程序
from fastapi import File, UploadFile
@app.post("/upload")
def upload(file: UploadFile = File(...)):
try:
contents = file.file.read()
with open(file.filename, 'wb') as f:
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
return {"message": f"Successfully uploaded {file.filename}"}
分块读取文件
如前所述，并在 本回答 ，如果文件太大而无法放入内存中（例如，如果您有 8GB 的 RAM，则无法加载 50GB 的文件（更不用说可用 RAM 总是小于您机器上安装的总容量，因为其他应用程序将使用部分 RAM）— 您应该将文件分块加载到内存中，然后一次处理一个块的数据。但是，此方法可能需要更长时间才能完成，具体取决于您选择的块大小 — 在以下示例中，块大小为 1024 * 1024 字节（即 1MB）。您可以根据需要调整块大小。
from fastapi import File, UploadFile
@app.post("/upload")
def upload(file: UploadFile = File(...)):
try:
with open(file.filename, 'wb') as f:
while contents := file.file.read(1024 * 1024):
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
return {"message": f"Successfully uploaded {file.filename}"}
另一个选择是使用 shutil.copyfileobj() ，它用于将 file-like 对象的内容复制到另一个 file-like 对象（也请查看 此答案 ）。默认情况下，数据以块的形式读取，默认缓冲区（块）大小为 1MB（即 1024 * 1024 字节）（对于 Windows）和其他平台的 64KB，如此处的源代码所示 。 您可以通过传递可选参数来指定缓冲区大小 length 。注意：如果 length 传递了负值，则将读取文件的整个内容 - 另 f.read() .copyfileobj() 在后台使用（如在 此处的 ）。
from fastapi import File, UploadFile
import shutil
@app.post("/upload")
def upload(file: UploadFile = File(...)):
try:
with open(file.filename, 'wb') as f:
shutil.copyfileobj(file.file, f)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
return {"message": f"Successfully uploaded {file.filename}"}
测试.py
import requests
url = 'http://127.0.0.1:8000/upload'
file = {'file': open('images/1.png', 'rb')}
resp = requests.post(url=url, files=file)
print(resp.json())
有关 HTML <form> 示例，请参见 此处 .
上传多个文件（列表）
应用程序
from fastapi import File, UploadFile
from typing import List
@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
contents = file.file.read()
with open(file.filename, 'wb') as f:
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
file.file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
分块读取文件
正如本答案前面所述，如果您预计某些文件相当大，但没有足够的 RAM 来容纳从头到尾的所有数据，那么您应该将文件分块加载到内存中，这样每次处理一个块的数据（注意：根据需要调整块大小，下面是 1024 * 1024 字节）。
from fastapi import File, UploadFile
from typing import List
@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
with open(file.filename, 'wb') as f:
while contents := file.file.read(1024 * 1024):
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
file.file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
或者使用 shutil.copyfileobj() ：
from fastapi import File, UploadFile
from typing import List
import shutil
@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
with open(file.filename, 'wb') as f:
shutil.copyfileobj(file.file, f)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
file.file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
测试.py
import requests
url = 'http://127.0.0.1:8000/upload'
files = [('files', open('images/1.png', 'rb')), ('files', open('images/2.png', 'rb'))]
resp = requests.post(url=url, files=files)
print(resp.json())
有关 HTML <form> 示例，请参见 此处 .
