如何在psycopg2中获取COPY请求中被拒绝行的日志？

这个问题我之前也碰到过，psycopg2默认不会主动把COPY操作里的跳过日志暴露出来，得靠自定义通知处理器来捕获PostgreSQL发送的NOTICE消息，具体做法如下：

PostgreSQL在执行带ON_ERROR IGNORE和LOG_VERBOSITY VERBOSE的COPY命令时，会把跳过行的信息以NOTICE级别的消息发送给客户端，而psycopg2默认只是忽略这些消息，所以我们需要给数据库连接绑定一个自定义的通知处理函数，来捕获并处理这些日志。

具体实现步骤 #

• 第一步：定义一个通知处理函数，用来接收并处理PostgreSQL发送的NOTICE消息。你可以选择直接打印日志，或者把日志保存到变量/文件中。
• 第二步：在建立数据库连接后，把这个处理函数绑定到连接的通知处理器上。
• 第三步：正常执行你的COPY操作，此时被跳过行的日志就会被捕获到。

完整示例代码 #

from psycopg2 import connect

# 定义自定义通知处理函数，这里我们把消息打印出来，也可以保存到变量/文件
def handle_notice(conn, notice):
    print(f"COPY日志: {notice.message.strip()}")

# 数据库连接配置（请替换成你的实际配置）
conn_config = {
    "dbname": "your_db",
    "user": "your_user",
    "password": "your_pw",
    "host": "your_host"
}

conn = connect(**conn_config)
# 绑定通知处理函数
conn.set_notice_processor(handle_notice)

curs = conn.cursor()
sql = """COPY PUBLIC.TEST_TABLE(
       TEST
       )
       FROM STDIN
       WITH (
       FORMAT CSV,
       ON_ERROR IGNORE,
       LOG_VERBOSITY VERBOSE 
       )"""

# 假设csv是你的输入文件对象
with open("your_data.csv", "r") as csv:
    curs.copy_expert(sql, csv)

conn.commit()
conn.close()

额外说明 #

• 如果需要把日志保存下来后续分析，可以修改handle_notice函数，把消息存入列表：

  copy_notices = []
  def handle_notice(conn, notice):
      copy_notices.append(notice.message.strip())
  

  执行完COPY操作后，copy_notices列表就会包含所有被跳过行的详细日志。
• 确保你的COPY语句中保留了LOG_VERBOSITY VERBOSE参数，这样才能获取到最详细的跳过原因（比如具体哪一行、哪个列、什么错误）。

备注：内容来源于stack exchange，提问作者Luice712