如何在 FastAPI 中保存 UploadFile

Question

我通过 POST 接受文件。当我保存到本地时，我可以使用file.read()读取内容，但是显示的是via file.name incorrect(16) 的名称。当我试图用这个名字找到它时，我得到了一个错误。可能是什么问题？

我的代码：

   @router.post(
    path="/upload",
    response_model=schema.ContentUploadedResponse,
)
async def upload_file(
        background_tasks: BackgroundTasks,
        uploaded_file: UploadFile = File(...)):
    uploaded_file.file.rollover()
    uploaded_file.file.flush()
    #shutil.copy(uploaded_file.file.name, f'../api/{uploaded_file.filename}')
    background_tasks.add_task(s3_upload, uploaded_file=fp)
    return schema.ContentUploadedResponse()

原文由 Fyzzys 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

背景

UploadFile 只是 SpooledTemporaryFile 的包装器，可以作为 UploadFile.file 访问。

SpooledTemporaryFile() […] 函数的 运行与 TemporaryFile() 完全相同

关于 TemporaryFile 文档 说：

返回一个可以用作临时存储区 的类文件 对象。 [..] 它会在关闭后立即销毁（包括对象被垃圾回收时的隐式关闭）。在 Unix 下，文件的目录项要么根本不创建，要么在文件创建后立即删除。其他平台不支持； 您的代码不应依赖于使用此函数创建的临时文件，该文件在文件系统中具有或不具有可见名称。

async def 端点

You should use the following async methods of UploadFile : write , read , seek and 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 端点

此外，我想引用本 主题 中的几个有用的实用程序函数（所有学分@dmontagu）使用 shutil.copyfileobj 和内部 UploadFile.file 。可以从 def 端点调用此函数：

 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。

原文由 alex_noname 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

您可以通过这种方式保存上传的文件，

 from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload-file/")
async def create_upload_file(uploaded_file: UploadFile = File(...)):
    file_location = f"files/{uploaded_file.filename}"
    with open(file_location, "wb+") as file_object:
        file_object.write(uploaded_file.file.read())
    return {"info": f"file '{uploaded_file.filename}' saved at '{file_location}'"}

您还可以使用 shutil.copyfileobj(...) 方法（请参阅 此详细答案 以了解两者在幕后的工作方式）。

因此，作为替代方法，您可以使用 shutil.copyfileobj(...) 编写类似下面的内容来实现文件上传功能。

 import shutil
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/upload-file/")
async def create_upload_file(uploaded_file: UploadFile = File(...)):
file_location = f"files/{uploaded_file.filename}"
    with open(file_location, "wb+") as file_object:
        shutil.copyfileobj(uploaded_file.file, file_object)
return {"info": f"file '{uploaded_file.filename}' saved at '{file_location}'"}

原文由 JPG 发布，翻译遵循 CC BY-SA 4.0 许可协议