首先,根据 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>
示例,请参见 此处 .