如何使用 FastAPI 上传文件？

2024-12-02 08:41:00
admin
原创
256
摘要：问题描述：我按照官方文档使用 FastAPI 上传文件，如下所示：@app.post("/create_file") async def create_file(file: UploadFile = File(...)): file2store = await file.rea...

问题描述：

我按照官方文档使用 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 文档，你需要安装python-multipart— 如果尚未安装 — 因为上传的文件将作为“表单数据”发送。例如：

pip install python-multipart

下面的示例使用对象.file的属性UploadFile来获取实际的 Python 文件（即SpooledTemporaryFile），这允许您调用SpooledTemporaryFile的方法，例如和.read()，.close()而不必await它们。 但是，在这种情况下，用定义端点非常重要def- 否则，如果端点是用定义的async def，则此类操作将阻止服务器，直到它们完成。 在 FastAPI 中，正常def端点在外部线程池中运行，然后await被编辑，而不是直接调用（因为它会阻止服务器）。 我强烈建议您看一下这个答案def，它解释了和端点之间的区别async def，并提供了在需要在端点内运行阻塞操作时的许多解决方案async def。

SpooledTemporaryFileFastAPI/Starlette 使用的属性max_size设置为 1 MB，这意味着数据在内存中缓冲，直到文件大小超过 1 MB，此时数据将写入磁盘上的临时文件（位于操作系统的临时目录下）。因此，如果您上传的文件大于 1 MB，则不会存储在内存中，调用file.file.read()实际上会将数据从磁盘读入内存。因此，如果文件太大而无法放入服务器的 RAM 中，您应该分块读取文件并一次处理一个块，如下面“分块读取文件”部分中所述。

如上所述以及在此答案def中也解释过，FastAPI / Starlette 使用 AnyIO 线程在外部线程池中运行阻塞函数（例如用 normal 定义的端点） ，然后运行await它们（以便 FastAPI 仍然可以异步工作），以防止它们阻塞事件循环（主线程的事件循环），从而阻塞整个服务器。因此，每次 HTTP 请求到达用 normal 定义的端点时def，都会生成一个新线程（或者如果可用，将使用一个空闲线程），因此，根据项目的要求、预期的流量（即同时访问您的 API 的用户数量），以及 API 中最终将在该线程池中运行的任何其他阻塞函数（有关更多详细信息，请参阅上面链接的答案），您可能需要调整该线程池中的最大线程数（有关如何执行此操作，请参阅上面链接的答案）。

但是，您应该始终尽可能使用异步代码（即使用async/ await），因为async代码直接在单个线程（在本例中为主线程）中运行的事件循环中运行。一种选择是定义端点async def并使用FastAPI 提供的异步 read()/ write()/ /etc. 文件方法，如此答案中所示。但是，您应该注意，正如此答案中所解释的那样，FastAPI 在后台实际上是在与前面描述的外部线程池不同的线程中调用相应的同步Python File 方法。因此，它可能会产生也可能不会产生很大的不同（在选择一种方法而不是另一种方法之前，请务必执行并比较测试）。close()

请注意，在本答案的底部以及本答案中，解释并演示了另一种方法，即如何使用 Starlette分块上传大文件request.stream()，这可以大大减少上传文件所需的时间，并避免使用该线程池中的线程。因此，我强烈建议您看一下。

上传单个文件

应用程序

from fastapi import File, UploadFile, HTTPException

@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:
        raise HTTPException(status_code=500, detail='Something went wrong')
    finally:
        file.file.close()

    return {"message": f"Successfully uploaded {file.filename}"}
分块读取文件

如前所述，以及在此答案中，如果文件太大而无法放入内存中（例如，如果您有 8GB 的​​ RAM，则无法加载 50GB 的文件（更不用说可用的 RAM 总是小于您机器上安装的总数量，因为其他应用程序将使用部分 RAM）——您应该将文件分块加载到内存中，然后一次处理一个块的数据。但是，此方法可能需要更长时间才能完成，具体取决于您选择的块大小——在下面的示例中，块大小为1024 * 1024字节（即 1MB）。您可以根据需要调整块大小。

from fastapi import File, UploadFile, HTTPException
        
@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:
        raise HTTPException(status_code=500, detail='Something went wrong')
    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, HTTPException
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:
        raise HTTPException(status_code=500, detail='Something went wrong')
    finally:
        file.file.close()
        
    return {"message": f"Successfully uploaded {file.filename}"}

test.py (使用requests)

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())

test.py (使用httpx)

import httpx

url = 'http://127.0.0.1:8000/upload'
file = {'file': open('images/1.png', 'rb')}
resp = httpx.post(url=url, files=file) 
print(resp.json())

有关 HTML<form>示例，请参见此处。

上传多个文件（列表）

应用程序

from fastapi import File, UploadFile, HTTPException
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:
            raise HTTPException(status_code=500, detail='Something went wrong')
        finally:
            file.file.close()

    return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
分块读取文件

正如本答案前面所述，如果您预计某些文件相当大，但没有足够的 RAM 来容纳从头到尾的所有数据，那么您应该将文件分块加载到内存中，这样每次处理一个块的数据（注意：根据需要调整块大小，下面是1024 * 1024字节）。

from fastapi import File, UploadFile, HTTPException
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:
            raise HTTPException(status_code=500, detail='Something went wrong')
        finally:
            file.file.close()
            
    return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}

或者使用shutil.copyfileobj()：

from fastapi import File, UploadFile, HTTPException
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:
            raise HTTPException(status_code=500, detail='Something went wrong')
        finally:
            file.file.close()

    return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}

test.py (使用requests)

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())

test.py (使用httpx)

import httpx

url = 'http://127.0.0.1:8000/upload'
files = [('files', open('images/1.png', 'rb')), ('files', open('images/2.png', 'rb'))]
resp = httpx.post(url=url, files=files) 
print(resp.json())

有关 HTML<form>示例，请参见此处。

解决方案 2：

@app.post("/create_file/")
async def image(image: UploadFile = File(...)):
    print(image.file)
    # print('../'+os.path.isdir(os.getcwd()+"images"),"*************")
    try:
        os.mkdir("images")
        print(os.getcwd())
    except Exception as e:
        print(e) 
    file_name = os.getcwd()+"/images/"+image.filename.replace(" ", "-")
    with open(file_name,'wb+') as f:
        f.write(image.file.read())
        f.close()
   file = jsonable_encoder({"imagePath":file_name})
   new_image = await add_image(file)
   return {"filename": new_image}
相关推荐
置顶 政府信创国产化的10大政策解读
政府信创国产化的10大政策解读
  政府信创国产化的10大政策解读一、信创国产化的背景与意义信创国产化，即信息技术应用创新国产化，是当前中国信息技术领域的一个重要发展方向。其核心在于通过自主研发和创新，实现信息技术应用的自主可控，减少对外部技术的依赖，并规避潜在的技术制裁和风险。随着全球信息技术竞争的加剧，以及某些国家对中国在科技领域的打压，信创国产化显...
工程项目管理   1565  
置顶 2025年20款好用的项目管理软件推荐，项目管理提效的20个工具和技巧
2025年20款好用的项目管理软件推荐，项目管理提效的20个工具和技巧
  为什么项目管理通常仍然耗时且低效？您是否还在反复更新电子表格、淹没在便利贴中并参加每周更新会议？这确实是耗费时间和精力。借助软件工具的帮助，您可以一目了然地全面了解您的项目。如今，国内外有足够多优秀的项目管理软件可以帮助您掌控每个项目。什么是项目管理软件？项目管理软件是广泛行业用于项目规划、资源分配和调度的软件。它使项...
项目管理软件   1354  
信创国产芯片如何推动自主可控生态建设
信创国产芯片如何推动自主可控生态建设
  信创国产芯片作为信息技术创新的核心领域，对于推动国家自主可控生态建设具有至关重要的意义。在全球科技竞争日益激烈的背景下，实现信息技术的自主可控，摆脱对国外技术的依赖，已成为保障国家信息安全和产业可持续发展的关键。国产芯片作为信创产业的基石，其发展水平直接影响着整个信创生态的构建与完善。通过不断提升国产芯片的技术实力、产...
国产信创系统   21  
信创生态建设中的5个关键挑战与解决方案
信创生态建设中的5个关键挑战与解决方案
  信创生态建设旨在实现信息技术领域的自主创新和安全可控，涵盖了从硬件到软件的全产业链。随着数字化转型的加速，信创生态建设的重要性日益凸显，它不仅关乎国家的信息安全，更是推动产业升级和经济高质量发展的关键力量。然而，在推进信创生态建设的过程中，面临着诸多复杂且严峻的挑战，需要深入剖析并寻找切实可行的解决方案。技术创新难题技...
信创操作系统   27  
信创国产芯片的研发难点与解决方案
信创国产芯片的研发难点与解决方案
  信创产业作为国家信息技术创新发展的重要领域，对于保障国家信息安全、推动产业升级具有关键意义。而国产芯片作为信创产业的核心基石，其研发进展备受关注。在信创国产芯片的研发征程中，面临着诸多复杂且艰巨的难点，这些难点犹如一道道关卡，阻碍着国产芯片的快速发展。然而，科研人员和相关企业并未退缩，积极探索并提出了一系列切实可行的解...
国产化替代产品目录   28  
热门文章
项目管理软件
政府信创国产化的10大政策解读
2025soft
2025年20款好用的项目管理软件推荐，项目管理提效的20个工具和技巧
img020
使用 Python 'Requests' 模块的代理
img044
10个项目管理常见问题解答：包括难点和解决方法
项目管理软件
项目延期的原因有哪些？如何有效预防和管理？
项目里程碑计划模板
项目里程碑计划模板怎么写?设定项目里程碑的5个方法
项目管理软件
项目管理指南：制定项目实施计划的6大步骤有哪些？
项目管理软件有哪些？
热门标签
更多>
IPD流程 项目管理 IT项目管理 敏捷开发 项目管理软件 Scrum 敏捷项目管理 IPD流程管理 项目风险管理 项目管理工具 项目经理 极限编程  产品管理  项目经理职责 项目路线图 项目进度管理 团队管理 测试管理工具 项目预算管理 PMP认证 瀑布式管理 项目范围管理 IT科技 项目管理平台 工程项目管理 项目资源管理 pmp证书考试
云禅道AD
禅道项目管理软件

云端的项目管理软件

尊享禅道项目软件收费版功能

无需维护，随时随地协同办公

内置subversion和git源码管理

每天备份，随时转为私有部署

免费试用