您可以使用同步写入编写文件（），在使用def定义端点后，在this answer中展示，或者使用异步编写(利用aiofiles)，在使用async def定义端点之后;UploadFile方法是async方法，因此需要await等待。以下是示例。有关def与async def以及它们如何影响API的性能(取决于端点内执行的任务)的更多详细信息，请查看this answer。
上传单个文件
app.py
from fastapi import File, UploadFile
import aiofiles
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
try:
contents = await file.read()
async with aiofiles.open(file.filename, 'wb') as f:
await f.write(contents)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
await file.close()
return {"message": f"Successfuly uploaded {file.filename}"}
分块读取文件
如此答案所解释的那样，FastAPI/Starlette在底层使用一个SpooledTemporaryFile，该文件的max_size属性设置为1 MB，这意味着数据会在内存中缓存，直到文件大小超过1 MB，此时数据将被写入磁盘上的临时文件，因此，调用await file.read()实际上会从磁盘读取数据到内存中（如果上传的文件大于1 MB）。因此，您可能希望以分块的方式使用async，以避免将整个文件加载到可能导致问题的内存中——例如，如果您有8GB的RAM，您无法加载50GB的文件（更不用说可用的RAM总是少于安装的总量，因为本机操作系统和其他运行在您的计算机上的应用程序将使用一些RAM）。因此，在这种情况下，您应该选择将文件分块加载到内存中，并逐个处理数据块。然而，这种方法可能需要更长的时间才能完成，具体取决于您选择的块大小；下面是1024 * 1024字节（= 1MB）。您可以根据需要调整块大小。
from fastapi import File, UploadFile
import aiofiles
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
try:
async with aiofiles.open(file.filename, 'wb') as f:
while contents := await file.read(1024 * 1024):
await f.write(contents)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
await file.close()
return {"message": f"Successfuly uploaded {file.filename}"}
或者，您可以使用{{link1：shutil.copyfileobj()}}，该函数用于将{{link2：类似文件}}对象的内容复制到另一个类似文件对象中（也可参见此答案）。默认情况下，数据会按块读入，Windows 的默认缓冲区（块）大小为 1MB（即 1024 * 1024 字节），其他平台为 64KB（请参见源代码{{link4：此处}}）。您可以通过传递可选的length参数来指定缓冲区大小。注意：如果传递负的length值，则会读取文件的全部内容，请参见{{link5：f.read()}}文档，该文档同样被.copyfileobj()所使用。您可以在{{link6：此处}}找到.copyfileobj()的源代码——在读写文件内容方面，它与先前的方法没有什么不同。然而，.copyfileobj()在后台使用阻塞I/O操作，如果在async def端点内使用，会导致整个服务器被阻塞。因此，为了避免这种情况，您可以使用Starlette的{{link7：run_in_threadpool()}}在单独的线程中运行所有所需的函数（然后等待），以确保主线程（其中协程运行）不被阻塞。当您调用UploadFile对象的async方法（如.write()、.read()、.close()等）时，FastAPI内部也会使用完全相同的函数，请参见源代码{{link8：此处}}。示例：
from fastapi import File, UploadFile
from fastapi.concurrency import run_in_threadpool
import shutil
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
try:
f = await run_in_threadpool(open, file.filename, 'wb')
await run_in_threadpool(shutil.copyfileobj, file.file, f)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
if 'f' in locals(): await run_in_threadpool(f.close)
await file.close()
return {"message": f"Successfuly uploaded {file.filename}"}
test.py
import requests
url = 'http://127.0.0.1:8000/upload'
file = {'file': open('images/1.png', 'rb')}
r = requests.post(url=url, files=file)
print(r.json())
关于HTML的<form>示例，请参见此处。
上传多个文件
app.py
from fastapi import File, UploadFile
import aiofiles
@app.post("/upload")
async def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
contents = await file.read()
async with aiofiles.open(file.filename, 'wb') as f:
await f.write(contents)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
await file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
分块读取文件
为了进行分块读取文件，请参见本答案中早先描述的方法。
test.py
import requests
url = 'http://127.0.0.1:8000/upload'
files = [('files', open('images/1.png', 'rb')), ('files', open('images/2.png', 'rb'))]
r = requests.post(url=url, files=files)
print(r.json())
有关 HTML 的 <form> 示例，请参见此处。
更新
深入研究源代码后发现，最新版本的
Starlette（FastAPI底层使用的框架）在
UploadFile数据结构中使用了一个带有
max_size属性设置为
1MB（
1024 * 1024字节）的
SpooledTemporaryFile - 可以在
这里看到 - 而旧版本中
max_size被设置为默认值，即0字节，例如
这里。
以上意味着，在过去，无论文件大小如何（如果文件无法适应RAM），数据都会完全加载到内存中，而在最新版本中，数据在内存中被分段缓冲，直到文件大小超过max_size（即1MB），此时内容将被写入磁盘；更具体地说，写入
操作系统的临时目录（注意：这也意味着您可以上传的文件的最大大小受限于系统
临时目录中可用的存储空间。如果您的系统有足够的存储空间（满足您的需求），则无需担心；否则，请查看
此答案以了解如何更改默认的临时目录）。因此，多次写入文件的过程——即最初将数据加载到RAM中，然后，如果数据超过1MB，则将文件写入临时目录，然后从临时目录读取文件（使用
file.read()），最后将文件写入永久目录——是使上传文件比使用Flask框架慢的原因，正如OP在他们的问题中指出的那样（尽管时间差异不大，但根据文件大小而定，只有几秒钟）。
解决方案
如果需要上传的文件超过1MB且上传时间很重要，则解决方案是将request body作为流访问。根据Starlette文档的说明，如果访问.stream()，则字节块会提供而不将整个body存储到内存中（如果body包含超过1MB的文件数据，则稍后会将其存储到临时目录中）。下面给出了一个示例，在客户端记录上传时间，最终与在OP问题中使用Flask框架的示例相同。
app.py
from fastapi import Request
import aiofiles
@app.post('/upload')
async def upload(request: Request):
try:
filename = request.headers['filename']
async with aiofiles.open(filename, 'wb') as f:
async for chunk in request.stream():
await f.write(chunk)
except Exception:
return {"message": "There was an error uploading the file"}
return {"message": f"Successfuly uploaded {filename}"}
如果您的应用程序不需要将文件保存到磁盘，而您只需要将文件直接加载到内存中，那么您可以使用以下代码（请确保您的RAM有足够的空间来容纳累积的数据）：
from fastapi import Request
@app.post('/upload')
async def upload(request: Request):
body = b''
try:
filename = request.headers['filename']
async for chunk in request.stream():
body += chunk
except Exception:
return {"message": "There was an error uploading the file"}
return {"message": f"Successfuly uploaded {filename}"}
test.py
import requests
import time
with open("images/1.png", "rb") as f:
data = f.read()
url = 'http://127.0.0.1:8000/upload'
headers = {'filename': '1.png'}
start = time.time()
r = requests.post(url=url, data=data, headers=headers)
end = time.time() - start
print(f'Elapsed time is {end} seconds.', '\n')
print(r.json())
如果您需要上传一个比较大的文件，而该文件无法适应客户端的RAM（例如，如果客户端设备上有2 GB可用RAM，并尝试加载4 GB的文件），则您应该在客户端使用流式上传，这将允许您发送大型流或文件而不将它们读入内存中（上传可能需要更长时间，具体取决于块大小，您可以通过分块读取文件并设置所需的块大小来自定义）。示例分别给出了Python requests和httpx（可能比requests具有更好的性能）。
test.py（使用requests）
import requests
import time
url = 'http://127.0.0.1:8000/upload'
headers = {'filename': '1.png'}
start = time.time()
with open("images/1.png", "rb") as f:
r = requests.post(url=url, data=f, headers=headers)
end = time.time() - start
print(f'Elapsed time is {end} seconds.', '\n')
print(r.json())
test.py（使用httpx）
import httpx
import time
url = 'http://127.0.0.1:8000/upload'
headers = {'filename': '1.png'}
start = time.time()
with open("images/1.png", "rb") as f:
r = httpx.post(url=url, data=f, headers=headers)
end = time.time() - start
print(f'Elapsed time is {end} seconds.', '\n')
print(r.json())
如需更多细节和代码示例（关于上传多个文件和基于上述方法（即使用request.stream()方法）的表单/JSON数据），请查看此答案。
