FastAPI端点上传Excel数据未更新SQLite数据库问题排查

核心问题定位 #

从代码中可直接发现参数匹配逻辑错误，这是数据库无更新的主要原因：

在FastAPI端点中，你已将Excel数据转换为字典列表data并传入db.uploadFile(name, data)；但在uploadFile方法中，处理字典列表的分支判断写为：

elif isinstance(data, list) and all(isinstance(item, dict) for item in data):

这里的data是方法内部初始化的空列表data = []，永远无法匹配传入的字典列表参数（方法第二个参数是file_path），导致该分支完全不执行，跳过所有数据插入逻辑。

排查思路 #

• 分支执行验证：在uploadFile各分支开头添加打印/日志，确认是否进入正确的字典列表处理分支
• 数据库事务检查：确认SQLite连接是否自动提交事务，with self.global_db是否真的会提交变更
• 异常捕获：检查是否有未捕获的异常导致流程中断，但端点仍返回成功信息
• 数据格式校验：验证转换后的字典列表键名是否与数据库表列名完全一致（含大小写、拼写）
• 表存在性确认：确认_taskboard_exists方法是否正确判断目标表存在，避免因表不存在导致静默失败

解决建议 #

1. 修复参数判断分支 #

修改uploadFile方法中的字典列表判断逻辑，将判断对象改为传入的file_path参数，并统一分支内变量名：

# 替换原有的elif分支
elif isinstance(file_path, list) and all(isinstance(item, dict) for item in file_path):
    # 处理字典列表数据
    upload_data = file_path
    if not upload_data:
        return  # 无数据可处理

    # 从第一个字典获取表头
    headers = list(upload_data[0].keys())

    # 确保表包含所有表头对应的列
    self._ensure_columns_exist(name, headers)

    with self.global_db:
        cursor = self.global_db.cursor()

        # 检查upload_id列是否存在
        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 not upload_id_exists:
            cursor.execute("ALTER TABLE '{}' ADD COLUMN upload_id INTEGER DEFAULT 0".format(name))
            upload_id = 0
        else:
            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)

        # 插入数据
        for row_dict in upload_data:
            values = [upload_id] + [row_dict.get(header) for header in headers]
            cursor.execute(
                "INSERT INTO '{}' ({}) VALUES ({})".format(
                    name, ",".join(["upload_id"] + headers), ",".join(["?"] * (len(headers) + 1))
                ),
                values
            )

注意去掉列名中的多余单引号（原代码"'upload_id'"会导致SQL语法错误，改为"upload_id"即可）。

2. 确保数据库事务提交 #

如果with self.global_db无法自动提交，在插入循环结束后手动添加提交语句：

cursor.execute("COMMIT;")

或检查global_db连接的autocommit属性是否设置为True。

3. 添加异常处理与日志 #

在FastAPI端点中添加异常捕获，避免静默失败：

@app.post("/upload/")
async def upload_file(name: str = Form(...), file: UploadFile = File(...)):
    try:
        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):
                    json_row[key] = value.isoformat()
                else:
                    json_row[key] = value
            return json_row

        data = [convert_to_json_serializable(row) for row in rows[1:]]

        db.uploadFile(name, data)
        return {"message": "Data uploaded successfully to Taskboard"}
    except Exception as e:
        return {"error": str(e)}, 400

在uploadFile关键步骤添加日志，方便排查：

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 进入字典列表分支时记录
logger.info(f"Processing dict list data for taskboard {name}")
# 插入完成后记录
logger.info(f"Inserted {len(upload_data)} rows into {name}")

4. 验证数据与SQL正确性 #

• 打印转换后的data，确认键名和值符合预期
• 打印最终执行的SQL语句和参数，检查是否存在语法错误或值不匹配情况

内容的提问来源于stack exchange，提问作者FFFFFFFFF