pythonpython-asynciotemporary-filesfastapi

How to save UploadFile in FastAPI


I accept the file via POST. When I save it locally, I can read the content using file.read (), but the name via file.name incorrect(16) is displayed. When I try to find it by this name, I get an error. What might be the problem?

My code:

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

Solution

  • Background

    UploadFile is just a wrapper around SpooledTemporaryFile, which can be accessed as UploadFile.file.

    SpooledTemporaryFile() [...] function operates exactly as TemporaryFile() does

    And documentation about TemporaryFile says:

    Return a file-like object that can be used as a temporary storage area. [..] It will be destroyed as soon as it is closed (including an implicit close when the object is garbage collected). Under Unix, the directory entry for the file is either not created at all or is removed immediately after the file is created. Other platforms do not support this; your code should not rely on a temporary file created using this function having or not having a visible name in the file system.

    async def endpoint

    You should use the following async methods of UploadFile: write, read, seek and close. They are executed in a thread pool and awaited asynchronously.

    For async writing files to disk you can use aiofiles. Example:

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

    Or in the chunked manner, so as not to load the entire file into memory:

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

    Also, I would like to cite several useful utility functions from this topic (all credits @dmontagu) using shutil.copyfileobj with internal UploadFile.file. This functions can be invoked from def endpoints:

    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
    

    Note: you'd want to use the above functions inside of def endpoints, not async def, since they make use of blocking APIs.