python 如何删除`return FileResponse(file_path)`后的文件

njthzxwz  于 2023-05-16  发布在  Python
关注(0)|答案(4)|浏览(149)

我使用FastAPI接收图像,处理它,然后将图像作为FileResponse返回。
但返回的文件是临时文件,需要在端点返回后删除。

@app.post("/send")
async def send(imagem_base64: str = Form(...)):

    # Convert to a Pillow image
    image = base64_to_image(imagem_base64)

    temp_file = tempfile.mkstemp(suffix = '.jpeg')
    image.save(temp_file, dpi=(600, 600), format='JPEG', subsampling=0, quality=85)

    return FileResponse(temp_file)

    # I need to remove my file after return it
    os.remove(temp_file)

我如何删除文件后返回它?

uwopmtnx

uwopmtnx1#

您可以删除background task中的文件,因为它将在 * 响应发送后 * 运行。

import os
import tempfile

from fastapi import FastAPI
from fastapi.responses import FileResponse

from starlette.background import BackgroundTasks

app = FastAPI()

def remove_file(path: str) -> None:
    os.unlink(path)

@app.post("/send")
async def send(background_tasks: BackgroundTasks):
    fd, path = tempfile.mkstemp(suffix='.txt')
    with os.fdopen(fd, 'w') as f:
        f.write('TEST\n')
    background_tasks.add_task(remove_file, path)
    return FileResponse(path)

另一种方法是使用dependency with yieldfinally块代码将在响应发送后执行,甚至在所有后台任务完成后执行。

import os
import tempfile

from fastapi import FastAPI, Depends
from fastapi.responses import FileResponse

app = FastAPI()

def create_temp_file():
    fd, path = tempfile.mkstemp(suffix='.txt')
    with os.fdopen(fd, 'w') as f:
        f.write('TEST\n')
    try:
        yield path
    finally:
        os.unlink(path)

@app.post("/send")
async def send(file_path=Depends(create_temp_file)):
    return FileResponse(file_path)
  • 注意 *:mkstemp()返回一个带有文件描述符和路径的元组。
t1qtbnec

t1qtbnec2#

可以将cleanup任务作为FileResponse的参数传递:

from starlette.background import BackgroundTask

# ...

def cleanup():
    os.remove(temp_file)

return FileResponse(
    temp_file,
    background=BackgroundTask(cleanup),
)

更新12-08-2022

如果某人动态地生成文件名,则可以将参数传递给后台任务,例如如下

return FileResponse(
    temp_file,
    background=BackgroundTask(cleanup, file_path),
)

然后需要调整cleanup函数以接受一个参数(将是文件名),并使用文件名作为参数而不是全局变量来调用os.remove函数

3lxsmp7m

3lxsmp7m3#

建议发送FileResponse并附加一个删除文件或文件夹的后台任务。
click here更多信息
后台任务将在响应完成后运行,因此它可以安全地删除文件/文件夹。

# ... other important imports
from starlette.background import BackgroundTasks

@app.post("/send")
async def send(imagem_base64: str = Form(...), bg_tasks: BackgroundTasks):

    # Convert to a Pillow image
    image = base64_to_image(imagem_base64)

    temp_file = tempfile.mkstemp(suffix = '.jpeg')
    image.save(temp_file, dpi=(600, 600), format='JPEG', subsampling=0, quality=85)

    bg_tasks.add_task(os.remove, temp_file)
 
   return FileResponse(
    temp_file,
    background=bg_tasks
   )
goqiplq2

goqiplq24#

为了确保在请求后删除临时文件,以及在请求中途意外失败时,建议使用FastAPI的Dependency Injection系统:

import os
from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

async def temp_path():
    """Create (and finally delete) a temporary file in a safe and non-blocking fashion."""
    loop = asyncio.get_running_loop()
    _, path = await loop.run_in_executor(None, tempfile.mkstemp)
    try:
        yield path
    finally:
        await loop.run_in_executor(None, os.unlink, path)

@app.get("/test")
async def test(
    ...,
    temp_path_1: Annotated[str, Depends(temp_path, use_cache=False)],
    temp_path_2: Annotated[str, Depends(temp_path, use_cache=False)],
):
    assert temp_path_1 != temp_path_2, "2 unique files due to use_cache=False"

    if "x" in temp_path_1:
        raise RuntimeError("Unexpected internal server error still deletes the files")

    return FileResponse(
        temp_path_1,
        media_type="video/mp4",
        filename="video_out.mp4",
    )

相关问题