背景
UploadFile 只是SpooledTemporaryFile 的包装，可以通过UploadFile.file 访问。
SpooledTemporaryFile() [...] 函数完全按照 TemporaryFile() 的方式操作
Given 为TemporaryFile：
返回一个类文件对象，可以用作临时存储区域。 [..] 它一关闭就会被销毁（包括对象被垃圾回收时的隐式关闭）。在 Unix 下，文件的目录条目要么根本不创建，要么在文件创建后立即删除。其他平台不支持； 您的代码不应依赖于使用此函数创建的临时文件，该文件在文件系统中具有或不具有可见名称。
async def端点
您应该使用UploadFile 中的以下异步methods：write、read、seek 和close。它们在线程池中执行并异步等待。
对于将文件异步写入磁盘，您可以使用aiofiles。示例：
@app.post("/")
async def post_endpoint(in_file: UploadFile=File(...)):
# ...
async with aiofiles.open(out_file_path, 'wb') as out_file:
content = await in_file.read() # async read
await out_file.write(content) # async write
return {"Result": "OK"}
或者分块的方式，以免把整个文件加载到内存中：
@app.post("/")
async def post_endpoint(in_file: UploadFile=File(...)):
# ...
async with aiofiles.open(out_file_path, 'wb') as out_file:
while content := await in_file.read(1024): # async read chunk
await out_file.write(content) # async write chunk
return {"Result": "OK"}
def端点
另外，我想从这个topic（所有学分@dmontagu）中引用几个有用的实用函数，使用shutil.copyfileobj和内部UploadFile.file：
import shutil
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Callable
from fastapi import UploadFile
def save_upload_file(upload_file: UploadFile, destination: Path) -> None:
try:
with destination.open("wb") as buffer:
shutil.copyfileobj(upload_file.file, buffer)
finally:
upload_file.file.close()
def save_upload_file_tmp(upload_file: UploadFile) -> Path:
try:
suffix = Path(upload_file.filename).suffix
with NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
shutil.copyfileobj(upload_file.file, tmp)
tmp_path = Path(tmp.name)
finally:
upload_file.file.close()
return tmp_path
def handle_upload_file(
upload_file: UploadFile, handler: Callable[[Path], None]
) -> None:
tmp_path = save_upload_file_tmp(upload_file)
try:
handler(tmp_path) # Do something with the saved temp file
finally:
tmp_path.unlink() # Delete the temp file
注意：您希望在 def 端点内使用上述函数，而不是 async def，因为它们使用阻塞 API。