如何在 FastAPI 中保存 UploadFile

Question

我通过 POST 接受该文件。本地保存时，可以使用file.read()读取内容，但是显示通过file.name不正确(16)的名称。当我尝试按此名称查找它时，出现错误。可能是什么问题？

我的代码：

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

Answer 1

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

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}'"}

Answer 2

背景

UploadFile只是一个包装器SpooledTemporaryFile，可以作为UploadFile.file.

SpooledTemporaryFile() [...] 函数的操作与 TemporaryFile()完全一样

鉴于为TemporaryFile：

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

async def 端点

您应该使用以下异步方法的UploadFile：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 端点

另外，我想从这个主题中引用几个有用的实用函数（所有学分@dmontagu），使用shutil.copyfileobjwith internal 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。