使用Psycopg2批量创建表并导入4000个CSV至PostgreSQL

嘿，我来帮你搞定这个批量CSV导入的问题——我之前处理过类似的大规模数据导入场景，咱们一步步拆解你的问题，给出实操性的方案：

1. 解决.format替换%s的冲突问题 #

你遇到的.format无法正确替换%s的问题，本质是把psycopg2的参数占位符和字符串格式化的替换符搞混了。psycopg2用%s作为列值的占位符，但如果你想用变量替换表名/列名这类标识符，不能直接用.format，否则会和占位符冲突，还可能引发SQL注入风险。

正确的做法是用psycopg2自带的sql模块来安全拼接标识符：

from psycopg2 import sql

# 假设你要动态指定表名sec_name
sec_name = "your_table"
cur.execute(
    sql.SQL("INSERT INTO {} (col1, col2) VALUES (%s, %s)").format(sql.Identifier(sec_name)),
    (val1, val2)  # 这里的%s对应列值的占位符，正常传参
)

你的Python3.5和PostgreSQL9.5环境完全支持这个模块（psycopg2 2.7+即可），这是最安全的方式，完全避免了格式化冲突和注入问题。

2. 选copy_from还是指定列插入？ #

毫无疑问，优先用copy_from！逐行插入（哪怕是批量executemany）的效率和copy_from根本不在一个量级——copy_from是PostgreSQL原生COPY命令的Python接口，能直接把文件内容批量写入数据库，速度快10~100倍，特别适合你4000个文件的场景。

关于列的指定：

• 如果你的CSV列顺序、数量和目标表完全一致，且后续不会变动，可以不指定columns参数；
• 但更稳妥的做法是显式指定columns，哪怕现在结构一致，后续表结构调整时能避免导入错误，比如：

cur.copy_from(csv_file, target_table, columns=["col1", "col2", "col3"], sep=',')

另外注意：如果你的CSV有表头，记得先跳过第一行（用next(f)），否则表头会被当成数据导入。

3. 简化并优化整个导入流程 #

结合你的环境（Python3.5 + PostgreSQL9.5），我整理了一套极简且高效的批量导入脚本，包含了上下文管理、错误处理和批量处理：

import psycopg2
import os
from psycopg2 import sql

# 配置参数
CSV_DIR = "/path/to/your/csv/folder"
TARGET_TABLE = "your_target_table"
COLUMNS = ["col1", "col2", "col3"]  # 替换成你的实际列名
DB_CONN_STR = "dbname=postgres user=postgres password=your_pwd host=localhost"

def batch_import_csv():
    # 用with语句自动管理连接和游标，避免资源泄漏
    with psycopg2.connect(DB_CONN_STR) as conn:
        for filename in os.listdir(CSV_DIR):
            if not filename.lower().endswith(".csv"):
                continue
            file_path = os.path.join(CSV_DIR, filename)
            print(f"开始处理文件: {filename}")
            
            try:
                with conn.cursor() as cur:
                    with open(file_path, 'r', encoding='utf-8') as f:
                        next(f)  # 跳过CSV表头（如果没有表头就删掉这行）
                        # 执行批量导入
                        cur.copy_from(
                            file=f,
                            table=sql.Identifier(TARGET_TABLE),
                            columns=COLUMNS,
                            sep=',',  # CSV分隔符，根据你的文件调整
                            null=''   # 把空字符串识别为NULL，按需调整
                        )
                conn.commit()
                print(f"✅ {filename} 导入成功")
            except Exception as e:
                conn.rollback()
                print(f"❌ {filename} 导入失败: {str(e)}")

if __name__ == "__main__":
    batch_import_csv()

优化点说明： #

• 用with语句管理数据库连接和游标，自动关闭资源，不用手动调用close()；
• 遍历目录自动过滤CSV文件，无需手动逐个处理；
• 每个文件单独提交事务，避免一个文件出错导致所有数据回滚；
• 加入错误捕获，方便排查导入失败的文件；
• 用sql.Identifier处理表名，安全避免SQL注入。

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