如何在Python导入CSV到PostgreSQL时添加文件名与时间戳

以下是几种实用的实现方案，根据你的数据量和场景选择：

方案一：临时表中转（大数据量首选） #

利用临时表先导入原始CSV数据，再批量插入到目标表时补充文件名和时间戳，保留copy_from的高效性：

import psycopg2
from datetime import datetime

# 连接数据库
conn = psycopg2.connect("dbname=你的数据库名 user=你的用户名")
cur = conn.cursor()
csv_filename = 'my.csv'
load_time = datetime.now()

# 创建与CSV列匹配的临时表（请替换为你的实际列名和类型）
cur.execute("""
CREATE TEMP TABLE temp_csv_import (
    col1 TEXT,
    col2 INT,
    col3 NUMERIC
)
""")

# 导入CSV到临时表
with open(csv_filename, 'r') as f:
    next(f)  # 跳过CSV表头
    cur.copy_from(f, 'temp_csv_import', sep=',')

# 批量插入到目标表，同时补充文件名和时间戳
cur.execute("""
INSERT INTO csv_import (col1, col2, col3, filename, load_timestamp)
SELECT col1, col2, col3, %s, %s
FROM temp_csv_import
""", (csv_filename, load_time))

# 提交并关闭连接
conn.commit()
cur.close()
conn.close()

方案二：Python预处理每行数据（小数据量直接用） #

直接读取CSV每行数据，追加文件名和时间戳后用executemany插入，逻辑简单直观：

import psycopg2
from datetime import datetime
import csv

conn = psycopg2.connect("dbname=你的数据库名 user=你的用户名")
cur = conn.cursor()
csv_filename = 'my.csv'
load_time = datetime.now()

# 读取并预处理CSV数据
with open(csv_filename, 'r') as f:
    reader = csv.reader(f)
    next(reader)  # 跳过表头
    # 给每行追加文件名和时间戳
    processed_data = [row + [csv_filename, load_time] for row in reader]

# 批量插入到目标表
cur.executemany("""
INSERT INTO csv_import (col1, col2, col3, filename, load_timestamp)
VALUES (%s, %s, %s, %s, %s)
""", processed_data)

conn.commit()
cur.close()
conn.close()

方案三：用StringIO模拟CSV文件（中等数据量兼顾效率） #

把预处理后的数据写入StringIO对象，再用copy_from导入，既保留COPY的高效性，又不用创建临时表：

import psycopg2
from datetime import datetime
import csv
from io import StringIO

conn = psycopg2.connect("dbname=你的数据库名 user=你的用户名")
cur = conn.cursor()
csv_filename = 'my.csv'
# 转成字符串格式，适配COPY的文本输入
load_time_str = datetime.now().isoformat()

output = StringIO()
writer = csv.writer(output, delimiter=',')

# 读取原始CSV并追加字段
with open(csv_filename, 'r') as f:
    reader = csv.reader(f)
    next(reader)  # 跳过表头
    for row in reader:
        row.append(csv_filename)
        row.append(load_time_str)
        writer.writerow(row)

# 回到模拟文件的开头
output.seek(0)
# 指定目标表的所有列，确保顺序匹配
cur.copy_from(
    output, 
    'csv_import', 
    sep=',', 
    columns=('col1', 'col2', 'col3', 'filename', 'load_timestamp')
)

conn.commit()
cur.close()
conn.close()

方案选择建议 #

• 数据量较大（10w行以上）：优先用临时表中转，COPY的效率远高于逐行插入
• 数据量较小（1w行以内）：用Python预处理每行，代码更简洁
• 中等数据量：用StringIO模拟文件，平衡效率和代码复杂度

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