Below are given various options on how to convert the uploaded .csv file into JSON. The following .csv sample file is used in the examples below.
data.csv
Id,name,age,height,weight
1,Alice,20,62,120.6
2,Freddie,21,74,190.6
3,Bob,17,68,120.0
Option 1
The csv.DictReader() method can accept as a file argument file objects as well. FastAPI’s UploadFile uses Python’s SpooledTemporaryFile, a file-like object (for more info on that, please have a look at this answer). You can access that through the .file attribute of the UploadFile object. However, since FastAPI/Starlette opens the file in bytes mode, if you passed it directly to the csv.DictReader() method, you would get an error, i.e., _csv.Error: iterator should return strings, not bytes. Hence, you could use codecs.iterdecode() (as suggested in this answer) that uses an incremental decoder to iteratively decode the input provided by the iterator (in this case from bytes to str). Example:
from fastapi import FastAPI, File, UploadFile
import csv
import codecs
app = FastAPI()
@app.post("/upload")
def upload(file: UploadFile = File(...)):
csvReader = csv.DictReader(codecs.iterdecode(file.file, 'utf-8'))
data = {}
for rows in csvReader:
key = rows['Id'] # Assuming a column named 'Id' to be the primary key
data[key] = rows
file.file.close()
return data
Output
{
"1": {
"Id": "1",
"name": "Alice",
"age": "20",
"height": "62",
"weight": "120.6"
},
"2": {
"Id": "2",
"name": "Freddie",
"age": "21",
"height": "74",
"weight": "190.6"
},
"3": {
"Id": "3",
"name": "Bob",
"age": "17",
"height": "68",
"weight": "120.0"
}
}
In case you wanted to return a list of dictionaries instead, you could use the below. Since the below would require the file to be open while returning the results, hence preventing the server from properly closing the file (by calling file.file.close()) when it’s done, one could use BackgroundTasks (which run after returning a response) to close the file:
from fastapi import FastAPI, File, UploadFile, BackgroundTasks
import csv
import codecs
app = FastAPI()
@app.post("/upload")
def upload(background_tasks: BackgroundTasks, file: UploadFile = File(...)):
csvReader = csv.DictReader(codecs.iterdecode(file.file, 'utf-8'))
background_tasks.add_task(file.file.close)
return list(csvReader)
Output
[
{
"Id": "1",
"name": "Alice",
"age": "20",
"height": "62",
"weight": "120.6"
},
{
"Id": "2",
"name": "Freddie",
"age": "21",
"height": "74",
"weight": "190.6"
},
{
"Id": "3",
"name": "Bob",
"age": "17",
"height": "68",
"weight": "120.0"
}
]
Option 2
Another solution would be to read the byte data of the uploaded file— using contents = file.file.read() (for async read/write see this answer)—then convert the bytes into string, and finally load them into an in-memory text buffer (i.e., StringIO), as mentioned here, which can be passed to csv.DictReader(). Example below:
from fastapi import FastAPI, File, UploadFile
import csv
from io import StringIO
app = FastAPI()
@app.post("/upload")
def upload(file: UploadFile = File(...)):
data = {}
contents = file.file.read()
buffer = StringIO(contents.decode('utf-8'))
csvReader = csv.DictReader(buffer)
for row in csvReader:
key = row['Id'] # Assuming a column named 'Id' to be the primary key
data[key] = row
buffer.close()
file.file.close()
return data
Option 3
To approach the problem in your way—i.e., using a filepath to read the csv file, instead of using the file contents directly or the file-like object, as described earler—you can copy the file contents into a NamedTemporaryFile, which unlike SpooledTemporaryFile that UploadFile provides, “has a visible name in the file system” that “can be used to open the file” (again, check this answer out for more info on that). Below is a working example:
from fastapi import FastAPI, File, UploadFile
from tempfile import NamedTemporaryFile
import os
import csv
app = FastAPI()
@app.post("/upload")
def upload(file: UploadFile = File(...)):
data = {}
temp = NamedTemporaryFile(delete=False)
try:
try:
contents = file.file.read()
with temp as f:
f.write(contents);
except Exception:
return {"message": "There was an error uploading the file"}
finally:
file.file.close()
with open(temp.name,'r', encoding='utf-8') as csvf:
csvReader = csv.DictReader(csvf)
for rows in csvReader:
key = rows['Id'] # Assuming a column named 'Id' to be the primary key
data[key] = rows
except Exception:
return {"message": "There was an error processing the file"}
finally:
#temp.close() # the `with` statement above takes care of closing the file
os.remove(temp.name) # Delete the file
return data
Option 4
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 (as described in this answer), you could convert the dataframe into a dictionary and return it—which, FastAPI, behind the scenes, will convert into JSON-compatible data, using the jsonable_encoder, and finally, serialise the data and return a JSONResponse (see this answer for more details). As a faster alternative, you could use the to_json() method and return a custom Response directly, as described in Option 1 (Update 2) of this answer.
from fastapi import FastAPI, File, UploadFile
from io import BytesIO
import pandas as pd
app = FastAPI()
@app.post("/upload")
def upload(file: UploadFile = File(...)):
contents = file.file.read()
buffer = BytesIO(contents)
df = pd.read_csv(buffer)
buffer.close()
file.file.close()
return df.to_dict(orient="records")
Note: If the file is too big and is taking up all of the memory and/or is taking too much time to process and/or return the results, please have a look at this answer, as well as this answer and this answer.