使用psycopg2导入带引号的逗号分隔CSV至PostgreSQL遇参数错误

问题场景 #

你有如下格式的CSV数据：

id, Name, Remarks, status
1, Juan, "Hello, World", active
2, Pepito, "Howdy, how are you", inactive

尝试用psycopg2的copy_from方法导入时，使用了以下代码：

import csv
import psycopg2

try:
    next(f)
    cursor.copy_from(f, table_name, sep=',', quotechar='"')
    conn.commit()
except (Exception, psycopg2.DatabaseError) as error:
    print("Error: %s" % error)
    conn.rollback()
    cursor.close()
    print(1)
print(f"Data successfully imported to {table_name}.")
cursor.close()

运行后出现错误：

Error: 'quotechar' is an invalid keyword argument for this function

期望导入后的表结构：

id|Name     |Remarks             |status
1 |Juan     |Hello, World        |active
2 |Pepito   |Howdy, how are you  |inactive

错误原因 #

psycopg2的cursor.copy_from()方法不支持quotechar参数，它仅适用于无引号包裹的简单分隔文本，无法处理包含引号的CSV数据。

解决方案 #

方法1：使用copy_expert()执行PostgreSQL原生COPY命令 #

PostgreSQL的COPY命令本身支持处理带引号的CSV，通过cursor.copy_expert()可以直接调用该命令，这是最高效的方式。

完整代码示例：

import psycopg2

# 建立数据库连接
conn = psycopg2.connect(
    dbname="your_db_name",
    user="your_username",
    password="your_password",
    host="your_host"
)
cursor = conn.cursor()
table_name = "target_table"
csv_file_path = "your_data.csv"

try:
    # 构造PostgreSQL COPY命令，指定分隔符、引号规则、跳过表头
    copy_command = f"""
        COPY {table_name} FROM STDIN WITH (
            DELIMITER ',',
            QUOTE '"',
            HEADER,
            FORMAT CSV,
            TRIM_WHITESPACE
        )
    """
    with open(csv_file_path, 'r', encoding='utf-8') as f:
        cursor.copy_expert(copy_command, f)
    conn.commit()
    print(f"Data successfully imported to {table_name}.")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"Error: {error}")
    conn.rollback()
finally:
    cursor.close()
    conn.close()

方法2：预处理CSV去除引号（备选） #

如果坚持使用copy_from，可以先用Python的csv.reader解析CSV，去除引号后写入临时文件再导入：

import csv
import psycopg2
import tempfile
import os

conn = psycopg2.connect(...)  # 填写你的数据库连接参数
cursor = conn.cursor()
table_name = "target_table"
csv_file_path = "your_data.csv"

try:
    with open(csv_file_path, 'r', encoding='utf-8') as infile, tempfile.NamedTemporaryFile(mode='w', delete=False) as outfile:
        reader = csv.reader(infile, quotechar='"')
        next(reader)  # 跳过表头行
        for row in reader:
            # 用逗号连接无引号的字段
            outfile.write(','.join(row) + '\n')
    # 用copy_from导入处理后的临时文件
    with open(outfile.name, 'r') as f:
        cursor.copy_from(f, table_name, sep=',')
    conn.commit()
    print(f"Data successfully imported to {table_name}.")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"Error: {error}")
    conn.rollback()
finally:
    cursor.close()
    conn.close()
    # 清理临时文件
    os.unlink(outfile.name)

推荐使用方法1，它直接利用PostgreSQL原生能力，性能更优且代码更简洁。

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