Python如何加载多个CSV文件到可迭代变量 逐行上传至SQL数据库

你的现有代码存在3个核心问题：

• csv_data未初始化就通过下标赋值，会直接抛出索引相关报错，需先将其定义为列表/字典类型
• csv.reader依赖活动的文件句柄运行，你在创建reader后立即关闭了CSV文件，后续读取时会因文件已关闭无法获取行数据
• 没有存储CSV与目标SQL表的映射关系，无法实现分表上传逻辑

实现方案 #

首先调整命令行传参格式，采用CSV文件路径:目标表名的规则传入参数，方便建立文件与表的对应关系，示例调用方式为python your_script.py ./user.csv:user_table ./order.csv:order_table

大文件友好的流式读取代码（无需全量加载CSV到内存） #

import csv
import sys
# 此处导入你使用的SQL连接库即可，比如pymysql、psycopg2等

def main():
    csv_tasks = []
    # 解析参数并初始化读取任务
    for arg in sys.argv[1:]:
        file_path, target_table = arg.split(':', 1)
        f = open(file_path, newline='', encoding='utf-8-sig')
        reader = csv.reader(f, dialect='excel', delimiter=',', quotechar='|')
        # 同时存储reader、目标表名、文件句柄
        csv_tasks.append({
            "reader": reader,
            "table": target_table,
            "file_handler": f
        })
    
    try:
        # 逐轮遍历取行，每轮取所有文件的当前行
        while True:
            all_file_end = True
            for task in csv_tasks:
                try:
                    current_row = next(task["reader"])
                    all_file_end = False
                    # 此处替换为你的SQL上传逻辑，task["table"]为目标表，current_row为当前行数据
                    # 示例伪代码：db_cursor.execute(f"INSERT INTO {task['table']} VALUES (%s,%s)", current_row)
                    print(f"已上传到表{task['table']}：{current_row}")
                except StopIteration:
                    # 当前文件已读完，直接跳过
                    continue
            if all_file_end:
                # 所有CSV都读取完毕，退出循环
                break
    finally:
        # 处理完所有数据后统一关闭文件句柄
        for task in csv_tasks:
            task["file_handler"].close()

if __name__ == "__main__":
    main()

小文件场景按指定行号读取代码（全量加载到内存） #

如果你的CSV文件体积不大，需要支持按任意行号读取，可以把所有行预先加载为列表：

# 初始化阶段替换为全量加载
csv_tasks.append({
    "rows": list(reader),
    "table": target_table,
    "file_handler": f
})

# 读取阶段按行号遍历
max_row_count = max(len(task["rows"]) for task in csv_tasks)
for row_idx in range(max_row_count):
    for task in csv_tasks:
        if row_idx < len(task["rows"]):
            current_row = task["rows"][row_idx]
            # 执行上传逻辑

可选配置 #

如果你的CSV第一行是表头不需要上传，在初始化reader后添加一行next(reader)即可跳过第一行。

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