时间戳

首页 > 解决方案 > 如何在 FastAPI 中保存 UploadFile

python - 如何在 FastAPI 中保存 UploadFile

问题描述

我通过 POST 接受文件。当我保存在本地时，我可以使用file.read()读取内容，但是显示通过file.name wrong(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()

标签： pythonpython-asynciotemporary-filesfastapi

解决方案

背景

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

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

给定TemporaryFile：

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

async def端点

您应该使用以下异步方法：、、和UploadFile。它们在线程池中执行并异步等待。writereadseekclose

对于将文件异步写入磁盘，您可以使用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。

推荐阅读