First, as per FastAPI documentation, you need to install python-multipart
—if you haven't already—as uploaded files are sent as "form data". For instance:
pip install python-multipart
The below examples use the .file
attribute of the UploadFile
object to get the actual Python file (i.e., SpooledTemporaryFile
), which allows you to call SpooledTemporaryFile
's methods, such as .read()
and .close()
, without having to await
them. It is important, however, to define your endpoint with def
in this case—otherwise, such operations would block the server until they are completed, if the endpoint was defined with async def
. In FastAPI, a normal def
endpoint is run in an external threadpool that is then awaited, instead of being called directly (as it would block the server).
The SpooledTemporaryFile
used by FastAPI/Starlette has the max_size
attribute set to 1 MB, meaning that the data are spooled in memory until the file size exceeds 1 MB, at which point the data are written to a temporary file on disk, under the OS's temp directory. Hence, if you uploaded a file larger than 1 MB, it wouldn't be stored in memory, and calling file.file.read()
would actually read the data from disk into memory. Thus, if the file is too large to fit into your server's RAM, you should rather read the file in chunks and process one chunk at a time, as described in "Read the File in chunks" section below. You may also have a look at this answer, which demonstrates another approach to upload a large file in chunks, using Starlette's .stream()
method and streaming-form-data
package that allows parsing streaming multipart/form-data
chunks, which results in considerably minimising the time required to upload files.
If you have to define your endpoint with async def
—as you might need to await
for some other coroutines inside your route—then you should rather use asynchronous reading and writing of the contents, as demonstrated in this answer. Moreover, if you need to send additional data (such as JSON
data) together with uploading the file(s), please have a look at this answer. I would also suggest you have a look at this answer, which explains the difference between def
and async def
endpoints.
Upload Single File
app.py
from fastapi import File, UploadFile
@app.post("/upload")
def upload(file: UploadFile = File(...)):
try:
contents = file.file.read()
with open(file.filename, 'wb') as f:
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
return {"message": f"Successfully uploaded {file.filename}"}
Read the File in chunks
As described earlier and in this answer, if the file is too big to fit into memory—for instance, if you have 8GB of RAM, you can't load a 50GB file (not to mention that the available RAM will always be less than the total amount installed on your machine, as other applications will be using some of the RAM)—you should rather load the file into memory in chunks and process the data one chunk at a time. This method, however, may take longer to complete, depending on the chunk size you choose—in the example below, the chunk size is 1024 * 1024
bytes (i.e., 1MB). You can adjust the chunk size as desired.
from fastapi import File, UploadFile
@app.post("/upload")
def upload(file: UploadFile = File(...)):
try:
with open(file.filename, 'wb') as f:
while contents := file.file.read(1024 * 1024):
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
return {"message": f"Successfully uploaded {file.filename}"}
Another option would be to use shutil.copyfileobj()
, which is used to copy the contents of a file-like
object to another file-like
object (have a look at this answer too). By default, the data is read in chunks with the default buffer (chunk) size being 1MB (i.e., 1024 * 1024
bytes) for Windows and 64KB for other platforms, as shown in the source code here. You can specify the buffer size by passing the optional length
parameter. Note: If negative length
value is passed, the entire contents of the file will be read instead—see f.read()
as well, which .copyfileobj()
uses under the hood (as can be seen in the source code here).
from fastapi import File, UploadFile
import shutil
@app.post("/upload")
def upload(file: UploadFile = File(...)):
try:
with open(file.filename, 'wb') as f:
shutil.copyfileobj(file.file, f)
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
return {"message": f"Successfully uploaded {file.filename}"}
test.py
import requests
url = 'http://127.0.0.1:8000/upload'
file = {'file': open('images/1.png', 'rb')}
resp = requests.post(url=url, files=file)
print(resp.json())
For an HTML <form>
example, see here.
Upload Multiple (List of) Files
app.py
from fastapi import File, UploadFile
from typing import List
@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
contents = file.file.read()
with open(file.filename, 'wb') as f:
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
file.file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
Read the Files in chunks
As described earlier in this answer, if you expect some rather large file(s) and don't have enough RAM to accommodate all the data from the beginning to the end, you should rather load the file into memory in chunks, thus processing the data one chunk at a time (Note: adjust the chunk size as desired, below that is 1024 * 1024
bytes).
from fastapi import File, UploadFile
from typing import List
@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
with open(file.filename, 'wb') as f:
while contents := file.file.read(1024 * 1024):
f.write(contents)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
file.file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
or, using shutil.copyfileobj()
:
from fastapi import File, UploadFile
from typing import List
import shutil
@app.post("/upload")
def upload(files: List[UploadFile] = File(...)):
for file in files:
try:
with open(file.filename, 'wb') as f:
shutil.copyfileobj(file.file, f)
except Exception:
return {"message": "There was an error uploading the file(s)"}
finally:
file.file.close()
return {"message": f"Successfuly uploaded {[file.filename for file in files]}"}
test.py
import requests
url = 'http://127.0.0.1:8000/upload'
files = [('files', open('images/1.png', 'rb')), ('files', open('images/2.png', 'rb'))]
resp = requests.post(url=url, files=files)
print(resp.json())
For an HTML <form>
example, see here.
python-multipart
installed? if it's not go forpip install python-multipart
f = open(file)
method once, and when I send the request several times, the f will be closed after the first time. Thank you agin for helping me.