close

How to save UploadFile in FastAPI

Hello Guys, How are you all? Hope You all Are Fine. Today We Are Going To learn about How to save UploadFile in FastAPI in Python. So Here I am Explain to you all the possible Methods here.

Without wasting your time, Let’s start This Article.

Table of Contents

How to save UploadFile in FastAPI?

  1. How to save UploadFile in FastAPI?

    In my case, I need to handle huge files, so I must avoid reading them all into memory. What I want is to save them to disk asynchronously, in chunks.

  2. save UploadFile in FastAPI

    In my case, I need to handle huge files, so I must avoid reading them all into memory. What I want is to save them to disk asynchronously, in chunks.

Method 1

You can save the uploaded files this way,

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

This is almost identical to the usage of shutil.copyfileobj(...) method.

So, the above function can be re-written as,

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

Method 2

In my case, I need to handle huge files, so I must avoid reading them all into memory. What I want is to save them to disk asynchronously, in chunks.

I’m experimenting with this and it seems to do the job (CHUNK_SIZE is quite arbitrarily chosen, further tests are needed to find an optimal size):

import os
import logging

from fastapi import FastAPI, BackgroundTasks, File, UploadFile

log = logging.getLogger(__name__)

app = FastAPI()

DESTINATION = "/"
CHUNK_SIZE = 2 ** 20  # 1MB


async def chunked_copy(src, dst):
    await src.seek(0)
    with open(dst, "wb") as buffer:
        while True:
            contents = await src.read(CHUNK_SIZE)
            if not contents:
                log.info(f"Src completely consumed\n")
                break
            log.info(f"Consumed {len(contents)} bytes from Src file\n")
            buffer.write(contents)


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    fullpath = os.path.join(DESTINATION, file.filename)
    await chunked_copy(file, fullpath)
    return {"File saved to disk at": fullpath}

However, I’m quickly realizing that create_upload_file is not invoked until the file has been completely received. So, if this code snippet is correct it will probably be beneficial to performance but will not enable anything like providing feedback to the client about the progress of the upload and it will perform a full data copy in the server. It seems silly to not be able to just access the original UploadFile temporary file, flush it and just move it somewhere else, thus avoiding a copy.

Summery

It’s all About this issue. Hope all Methods helped you a lot. Comment below Your thoughts and your queries. Also, Comment below which Method worked for you? Thank You.

Also, Read