How to delete the file after a `return FileResponse(file_path)`

Question

I'm using FastAPI to receive an image, process it and then return the image as a FileResponse.

But the returned file is a temporary one that need to be deleted after the endpoint return it.

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

How can I delete the file after return it ?

Answer 1

You can delete a file in a background task, as it will run after the response is sent.

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)

Another approach is to use dependency with yield. The finally block code will be executed after the response is sent and even after all background tasks have been completed.

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)

Note: mkstemp() returns a tuple with a file descriptor and a path.

Answer 2

You can pass the cleanup task as a parameter of FileResponse:

from starlette.background import BackgroundTask

# ...

def cleanup():
    os.remove(temp_file)

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

UPDATE 12-08-2022

If someone is generating the filename dynamically, then one may pass the parameters to the background task, e.g., as follows

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

The cleanup function then needs to be adapted to accept a parameter, which will be the filename, and call the os.remove function with the filename as parameter instead of the global variable

Answer 3

It is recommended to send FileResponse with a background task attached that deletes the file or folder.

click here for more info

The background task will run after the response has been served, so it can safely delete the file/folder.

# ... 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
   )    

Answer 4

To ensure the temporary file is deleted after the request, also when the request fails unexpectedly mid-flight, using FastAPI's Dependency Injection system is recommended:

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",
    )

Answer 5

I think this will work also

import tempfile
from fastapi import FileResponse


class TempFileResponse(FileResponse):
    def __init__(self, prefix, **params) -> None:
        self.temp_file = tempfile.NamedTemporaryFile(prefix=prefix)
        super().__init__(path=self.temp_file.name, **params)

    def __del__(self):
        # This will delete the file
        self.temp_file.close()


@router.get("/produce-data", response_class=FileResponse)
async def produce() -> FileResponse:
    file_name = "some_file_data.txt"
    logger.info(f"Downloading data as {file_name}")
    response_file = TempFileResponse(prefix="some_file_", filename=file_name)
    with open(response_file.temp_file.name, "w") as f:
        f.write("Hello, world!")
    return response_file