如何使用 FastAPI 上传文件？-rest-IT问答社区-解决你的IT疑问

如何使用 FastAPI 上传文件？

Regestea 1月前

我按照官方文档使用FastAPI上传文件，如下所示：@app.post(\'/create_file\')async def create_file(file: UploadFile = File(...)): file2stor...

我按照官方文档使用 FastAPI 上传文件，如下所示：

@app.post("/create_file")
async def create_file(file: UploadFile = File(...)):
      file2store = await file.read()
      # some code to store the BytesIO(file2store) to the other database

当我使用 Python 请求库发送请求时，如下所示：

f = open(".../file.txt", 'rb')
files = {"file": (f.name, f, "multipart/form-data")}
requests.post(url="SERVER_URL/create_file", files=files)

该 file2store 变量始终为空。有时（很少见），它可以获取文件字节，但几乎所有时间它都是空的，所以我无法在其他数据库上恢复该文件。

我也尝试了 bytes 而不是 UploadFile ，但得到的结果相同。我的代码有问题，还是我使用 FastAPI 上传文件的方式错误？

帖子版权声明 1、本帖标题：如何使用 FastAPI 上传文件？
本站网址：http://xjnalaquan.com/
2、本网站的资源部分来源于网络，如有侵权，请联系站长进行删除处理。
3、会员发帖仅代表会员个人观点，并不代表本站赞同其观点和对其真实性负责。
4、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报
5、站长邮箱：yeweds@126.com 除非注明，本帖由Regestea在本站《rest》版块原创发布，转载请注明出处！

最新回复 (0)

最新倒序只看楼主

cbdeveloper 1月前 0 取消查看

引用 3楼
首先，根据 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> 示例，请参见此处 .

关于作者

Regestea

UID:35794一级用户组

主题数
0

帖子数
0

精华数
0

注册排名
35794

导航

论坛

我的

如何使用 FastAPI 上传文件？

上传单个文件

分块读取文件

上传多个文件（列表）

分块读取文件

Regestea

TAG

作者主题

作者最近主题：