导入CSV到PostgreSQL时遭遇psycopg2.errors.UndefinedColumn错误

我有一个从CSV文件生成的pandas DataFrame，希望将其导入PostgreSQL数据库。实现代码如下：

import pandas as pd
import psycopg2

# Import CSV, create Data Frame
data = pd.read_csv('my_csv.csv', delimiter=';')
df = pd.DataFrame(data)

# Prepare Data (Rename / Shorten Column-Headers)
columnsFromCSV = list(df.columns)
for i in columnsFromCSV:
    columnName = i.rsplit(None, 2)[0]
    df.rename(columns={i : columnName}, inplace=True)
df.columns = df.columns.str.lower()

# Connect to Database
conn = psycopg2.connect(
   database='database', user='postgres', password='admin', host='127.0.0.1', port= '5432'
)
cursor = conn.cursor() 

# Insert Data Frame into Database
for i in df.columns[1:]:
    cursor.execute('INSERT INTO counter (counterid) VALUES ({0})'.format(i))
    for j in range(365):
        cursor.execute('INSERT INTO measurements (counterid) VALUES({0})'.format(i))
conn.commit()
conn.close()

需求是：先将每个计数器信息存入counter表一次，再为每个计数器生成365条measurements表的记录（每日测量值后续补充）。但执行代码时出现错误：

Traceback (most recent call last):
  File "c:\EnergyCounter\EnergyCounter\backend\CSV_read_script.py", line 25, in <module>
    cursor.execute('INSERT INTO counter (counterid) VALUES ({0})'.format(i))
psycopg2.errors.UndefinedColumn: ERROR:  Column »counter1« does not exist 
LINE 1: INSERT INTO counter (counterid) VALUES (counter1)

尝试过小写处理、引号包裹表名、使用%s替代.format等方法，但问题仍未解决。

错误根源 #

你用字符串格式化直接把计数器ID（比如counter1）拼进SQL语句，PostgreSQL会把它解析成列名而非字符串值，所以抛出“列不存在”的错误。psycopg2要求用参数化查询传递值，不能直接拼字符串。

修复后的代码 #

import pandas as pd
import psycopg2

# 读取CSV并处理列名
data = pd.read_csv('my_csv.csv', delimiter=';')
df = pd.DataFrame(data)

# 重命名列名并转为小写
for col in df.columns:
    new_col = col.rsplit(None, 2)[0]
    df.rename(columns={col: new_col}, inplace=True)
df.columns = df.columns.str.lower()

# 连接数据库
conn = psycopg2.connect(
    database='database', user='postgres', password='admin', host='127.0.0.1', port='5432'
)
cursor = conn.cursor()

try:
    # 插入计数器到counter表
    for counter_id in df.columns[1:]:
        # 使用参数化查询，%s是psycopg2的占位符
        cursor.execute('INSERT INTO counter (counterid) VALUES (%s)', (counter_id,))
        
        # 为每个计数器生成365条measurements记录
        # 用executemany批量插入，比循环单条效率高
        measurements_data = [(counter_id,) for _ in range(365)]
        cursor.executemany('INSERT INTO measurements (counterid) VALUES (%s)', measurements_data)
    
    conn.commit()
except Exception as e:
    conn.rollback()
    print(f"插入失败: {e}")
finally:
    # 确保关闭连接
    cursor.close()
    conn.close()

关键改进点 #

• 参数化查询：用%s作为占位符，把counter_id作为参数传入cursor.execute()，psycopg2会自动处理字符串转义，避免SQL注入和语法错误。
• 批量插入优化：用executemany()批量插入365条记录，比循环调用execute()效率高很多。
• 异常处理：增加事务回滚逻辑，避免部分插入成功导致数据不一致。
• 资源释放：确保游标和连接最终关闭，避免资源泄漏。

额外建议 #

如果你的counter表有主键约束（比如counterid是主键），需要先判断计数器是否已存在，避免重复插入：

cursor.execute('SELECT 1 FROM counter WHERE counterid = %s', (counter_id,))
if not cursor.fetchone():
    cursor.execute('INSERT INTO counter (counterid) VALUES (%s)', (counter_id,))

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