求助：如何用FastAPI和Pandas将CSV逐行导入MongoDB

问题分析 #

第一段代码失败原因 #

• df.to_dict(orient='records')生成了每行对应字典的列表，但未将结果赋值给变量，等于无效处理
• 最后调用insert_one(file.file)是插入文件对象而非转换后的数据集，必然无法成功插入数据

第二段代码的问题 #

• 把所有CSV行存入单个data字典，再用insert_one插入，导致所有数据变成MongoDB里的单个文档
• 应将每行的字典收集为列表，用insert_many批量插入实现每行对应独立文档

解决方案一：使用Pandas实现批量插入 #

利用Pandas快速转换CSV格式，再批量插入：

@router.post("/uploadfile", response_description="upload a file")
async def upload(request: Request, file: UploadFile = File(...)):
    contents = file.file.read()
    buffer = BytesIO(contents)
    # 可根据实际CSV格式调整sep、encoding等参数
    df = pd.read_csv(buffer)
    buffer.close()
    file.file.close()
    # 转换为每行对应字典的列表
    data_list = df.to_dict(orient='records')
    # 批量插入所有文档
    if data_list:
        await request.app.mongodb["SentenceModel"].insert_many(data_list)
    return {"message": f"成功插入{len(data_list)}个文档"}

解决方案二：使用标准csv模块实现批量插入 #

无需依赖第三方库，直接用Python标准库处理：

import csv
import codecs
from fastapi import UploadFile, File, Request

@router.post("/uploadfile", response_description="upload a file")
async def upload(request: Request, file: UploadFile = File(...)):
    csvReader = csv.DictReader(codecs.iterdecode(file.file, 'utf-8'))
    # 收集每行数据为字典列表
    data_list = [row for row in csvReader]
    file.file.close()
    if data_list:
        await request.app.mongodb["SentenceModel"].insert_many(data_list)
    return {"message": f"成功插入{len(data_list)}个文档"}

注意事项 #

• 确保CSV表头与MongoDB集合字段匹配，可在插入前根据需求调整字段名
• 若CSV数据量极大（如超10万行），建议分批次插入，避免内存溢出
• 可添加异常处理逻辑，捕获文件读取、数据库插入错误，返回更友好的响应

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