UploadFile uses Python's SpooledTemporaryFile, a "file stored in memory", and which "is destroyed as soon as it is closed". For more info on that, please have a look at this answer.
Option 1
To approach the problem on your way (i.e., reading from csv file and not using the file contents that you can get from contents = await file.read()), you can copy the file contents into a NamedTemporaryFile (again, check this answer out for more info on that), and then use it to iterate over the csv contents. Below is a working example:
import uvicorn
from fastapi import FastAPI, File, UploadFile
from tempfile import NamedTemporaryFile
import os
import csv
app = FastAPI()
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
contents = await file.read()
data = {}
file_copy = NamedTemporaryFile(delete=False)
try:
with file_copy as f: # The 'with' block ensures that the file closes and data are stored
f.write(contents);
with open(file_copy.name,'r', encoding='utf-8') as csvf:
csvReader = csv.DictReader(csvf)
for rows in csvReader:
key = rows['No']
data[key] = rows
finally:
file_copy.close() # Remember to close any file instances before removing the temp file
os.unlink(file_copy.name) # delete the file
return data
Option 2
Alternatively, a much more elegant solution would be to use, as mentioned earlier, the byte data of the uploaded file, saving you from copying them into a new temporary file. Convert the bytes into a string, and then load the string object into an in-memory text buffer (i.e., StringIO), as mentioned here, which you can pass to the csv reader. Example below:
from fastapi import FastAPI, File, UploadFile
import csv
from io import StringIO
app = FastAPI()
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
data = {}
contents = await file.read()
decoded = contents.decode()
buffer = StringIO(decoded)
csvReader = csv.DictReader(buffer)
for rows in csvReader:
key = rows['No']
data[key] = rows
buffer.close()
return data
Option 3
You could also write the bytes from the uploaded file to a BytesIO stream, which you could then convert into a pandas dataframe. Next, using the to_dict() method, you could convert the dataframe into dictionary and return it (which, by default, FastAPI will convert into JSON using the jsonable_encoder and return a JSONResponse).
from fastapi import FastAPI, File, UploadFile
from io import BytesIO
import pandas as pd
app = FastAPI()
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
contents = await file.read()
buffer = BytesIO(contents)
df = pd.read_csv(buffer)
buffer.close()
return df.to_dict(orient='records')
