Python/psycopg2 中优雅的主键错误处理

Question

我在使用 Python 2.7 和

在 [150]: psycopg2.version 输出为 '2.4.2 (dt dec pq3 ext)'

我有一个简单的 Python 脚本，用来处理交易并将数据写入数据库。偶尔会有一个插入操作违反了我的主键限制。这没关系，我只想让它忽略这个记录，继续执行后面的操作。问题是，psycopg2 的主键错误会导致整个事务块中断，之后的所有插入操作都失败。下面是一个示例错误：

ERROR: duplicate key value violates unique constraint "encounter_id_pkey"
DETAIL:  Key (encounter_id)=(9012235) already exists.

这个错误发生在下一个插入操作上，并不是违反主键。

Inserting: 0163168~9024065
ERROR: current transaction is aborted, commands ignored until end of transaction block

第二个错误会在每次插入时重复出现。这里有一个简化的循环。我是在遍历一个 pandas 数据框，但这可以是任何循环。

conn = psycopg2.connect("dbname='XXXX' user='XXXXX' host='XXXX' password='XXXXX'")

cur = conn.cursor()

for i, val in df2.iteritems():
    try:
        cur = conn.cursor()
        cur.execute("""insert into encounter_id_table (
        encounter_id,current_date  )
        values       
        (%(create_date)s, %(encounter_id)s ) ;""",
        'encounter_id':i.split('~')[1],  
        'create_date': datetime.date.today() })           
        cur.commit()
        cur.close() 
    except Exception , e:
        print 'ERROR:', e[0]
        cur.close()
 conn.close()   

基本的想法是优雅地处理错误。就像皇家海军的纳尔逊海军上将所说的：“去他的战术，直接冲上去。”在我们的情况下就是“去他的错误，直接处理它。”我以为每次插入时打开一个游标可以重置事务块。我不想因为一个主键错误就必须重置连接。我是不是漏掉了什么？

提前感谢你的时间。

约翰

Answer 1

首先，CURRENT_DATE 是在所有SQL标准和PostgreSQL中都被保留的关键词。你不能直接把它当作标识符使用，除非用双引号括起来。我更倾向于避免这样，所以在我的例子中我用了列名 curdate。

接下来，你似乎把插入列的顺序搞反了：

(%(create_date)s, %(encounter_id)s )

应该是：

( %(encounter_id)s, %(create_date)s)

关于你的主要问题：干脆避免这个问题吧！

自从Postgres 9.5版本以来，INSERT ... ON CONFLICT DO NOTHING 是解决这个问题的好方法。这也能避免在同时写入时出现的竞争条件：

INSERT INTO encounter_id_table (encounter_id, curdate)
VALUES (1234, CURRENT_DATE)
ON CONFLICT DO NOTHING;

可以参考：

• 如何在PostgreSQL中进行UPSERT（合并、插入...在重复时更新）？
• 如何在PostgreSQL中使用RETURNING与ON CONFLICT？

在Python语法中，应该是：

cur.execute("""INSERT INTO encounter_id_table (encounter_id, curdate)
    VALUES (%(encounter_id)s, CURRENT_DATE);""",
  {'encounter_id':i.split('~')[1]})       

在旧版本中，你可以大部分通过在插入之前检查键是否已经在表中来避免这个问题：

INSERT INTO encounter_id_table (encounter_id, curdate)
SELECT 1234, now()::date
WHERE  NOT EXISTS (SELECT FROM encounter_id_table t
                   WHERE t.encounter_id = 1234);

不过，这在同时写入时可能会失败。

Answer 2

如果出现错误，你应该撤销事务。

我在下面的代码中添加了一个额外的 try..except..else 结构，这样可以清楚地显示出异常会发生的具体位置。

try:
    cur = conn.cursor()

    try:
        cur.execute("""insert into encounter_id_table (
            encounter_id,current_date  )
            values       
            (%(create_date)s, %(encounter_id)s ) ;""",
            'encounter_id':i.split('~')[1],  
            'create_date': datetime.date.today() })
    except psycopg2.IntegrityError:
        conn.rollback()
    else:
        conn.commit()

    cur.close() 
except Exception , e:
    print 'ERROR:', e[0]