I’m working on a FastAPI project with a SQLite database where I have a feature to upload .xlsx files to update taskboards. Each taskboard is stored as a separate table in the SQLite database. My current implementation replaces the old data with the new data upon upload, but I want to modify it so that the new data is appended to the existing data instead.
Here’s the relevant part of my code:
The endpoint:
@app.post("/upload/")
async def upload_file(name: str = Form(...), file: UploadFile = File(...)):
contents = await file.read()
workbook = load_workbook(filename=BytesIO(contents))
sheet = workbook.active
rows = list(sheet.iter_rows(values_only=True))
headers = rows[0]
def convert_to_json_serializable(row):
json_row = {}
for key, value in zip(headers, row):
if isinstance(value, datetime):
# Convert datetime to ISO string format
json_row[key] = value.isoformat()
else:
json_row[key] = value
return json_row
data = [convert_to_json_serializable(row) for row in rows[1:]] # Convert rows to list of dicts
return data # Return the data as a list of dictionaries
The function for uploading:
def uploadFile(self, name, file_path):
# Check if the Taskboard exists
if not self._taskboard_exists(name):
raise Exception(f"Taskboard '{name}' does not exist.")
# Load data from the spreadsheet or CSV file
data = []
headers = None # To store column headers
if isinstance(file_path, str):
if file_path.endswith(".xlsx"):
wb = openpyxl.load_workbook(file_path)
ws = wb.active
headers = [
cell.value for cell in next(ws.iter_rows())
] # Corrected line
for row in ws.iter_rows(values_only=True):
data.append(row)
elif file_path.endswith(".csv"):
with open(file_path, "r") as csv_file:
csv_reader = csv.reader(csv_file)
headers = next(csv_reader)
for row in csv_reader:
data.append(row)
elif isinstance(file_path, bytes): # If file_path is bytes
# Convert bytes to BytesIO for openpyxl
inbytes = BytesIO(file_path)
wb = openpyxl.load_workbook(inbytes)
ws = wb.active
headers = [cell.value for cell in next(
ws.iter_rows())] # Corrected line
for row in ws.iter_rows(values_only=True):
data.append(row)
else:
raise Exception(
"Unsupported file format. Must be filepath as str or file as bytes",
type(file_path),
)
# Ensure that the Taskboard table has columns for all headers
self._ensure_columns_exist(name, headers)
# Prevent the headers from being reinserted, this generally causes errors
data.pop(0)
# Single quote names to prevent parsing errors when spaces are present in name
headers = list(map(lambda header: "'" + header + "'", headers))
# Insert data into the Taskboard table, setting missing columns to None
with self.global_db:
cursor = self.global_db.cursor()
# Check if an 'upload_id' column exists in the table
cursor.execute("PRAGMA table_info('{}')".format(name))
columns_info = cursor.fetchall()
upload_id_exists = any(
column[1] == "upload_id" for column in columns_info)
# If 'upload_id' column does not exist, add it with a default value of 0
if not upload_id_exists:
cursor.execute(
"ALTER TABLE '{}' ADD COLUMN 'upload_id' INTEGER DEFAULT 0".format(
name
)
)
upload_id = 0
else:
# Check the last 'upload_id' in the table
upload_id = cursor.execute(
"SELECT MAX(upload_id) FROM '{}'".format(name)
).fetchone()[0]
upload_id = (
0 if upload_id is None else (upload_id + 1)
) # Check if upload id has no entries.
# Insert upload_id into headers list
headers.insert(0, "upload_id")
for row in data:
# Add the 'upload_id' as the first column
values = [upload_id] + [
value if value is not None else None for value in row
]
cursor.execute(
"INSERT INTO '{}' ({}) VALUES ({})".format(
name,
",".join(headers),
# Add 1 for 'upload_id'
",".join(["?"] * len(headers)),
),
values,
)
I’m unsure how to modify the SQL queries to achieve this. Can someone guide me on how to adjust the method to append new data to the existing table?
Additional Context (Optional):
Each taskboard is represented as a separate table in the SQLite database.
The frontend is a JavaScript-based web application that sends .xlsx files to the FastAPI backend.
FastAPI handles the file and updates the SQLite database.
Any insights or code examples would be greatly appreciated!
