pysqlite的IntegrityError: 区分'NOT NULL'和'UNIQUE'违规

Question

在pysqlite中，如果违反了NOT NULL或UNIQUE的限制，就会出现一个叫做IntegrityError的错误。可惜的是，这种错误类型没有提供错误代码，只有一条信息。

假设我想忽略唯一性约束的违规，因为我知道在给定的数据下这样做是安全的，但如果关键列中有空值，我希望能够报告出来。

我想出了以下解决方案：

con = sqlite3.connect(':MEMORY:')
con.execute('''CREATE TABLE ABCD (A TEXT NOT NULL,
                                  B TEXT NOT NULL,
                                  C TEXT NOT NULL,
                                  D TEXT NOT NULL,
                                  PRIMARY KEY (A, B))''')
with con:
    for a, b, c, d in inputs:
        try:
            con.execute('INSERT INTO ABCD VALUES (?, ?, ?, ?)',
                        (a, b, c, d))
        except sqlite3.IntegrityError as e:
            # Skip 'not unique' errors, but raise others.
            if not e.message.endswith('not unique'):
                raise
con.close()

不过，解析错误信息似乎不太靠谱，可能会不可靠。有没有更好的方法来做到这一点，甚至可以使用con.executemany()？

Answer 1

下面是一个可以正常运行的代码：

import sqlite3

con = sqlite3.connect(':memory:')
con.execute('''CREATE TABLE ABCD (A TEXT NOT NULL,
                                  B TEXT NOT NULL,
                                  C TEXT NOT NULL,
                                  D TEXT NOT NULL,
                                  PRIMARY KEY (A, B));''')

inputs = [('cow', 'pig', 'cat', 'dog'), ('cow', 'pig', 'quail', 'turkey')]
with con:
    for a, b, c, d in inputs:
        try:
            con.execute('INSERT INTO ABCD VALUES (?, ?, ?, ?);',
                        (a, b, c, d))
        except sqlite3.IntegrityError as e:
            if 'not null' in e.args[0].lower():
                print('There is a NULL value')
            elif 'unique constraint' in e.args[0].lower():
                print('There is unique violation')
            else:
                raise

测试：

>>> 
There is a NULL value
>>> 

第二次测试的结果：

>>> 
There is unique violation
>>> 

希望这能对你有所帮助。

Answer 2

一个更优雅的解决方案是完全依赖SQL(ite)的功能。通过为主键指定一个冲突处理规则（ON CONFLICT IGNORE），我们就能实现想要的效果：

con = sqlite3.connect(':memory:')
con.execute('''CREATE TABLE ABCD (A TEXT NOT NULL,
                                  B TEXT NOT NULL,
                                  C TEXT NOT NULL,
                                  D TEXT NOT NULL,
                                  PRIMARY KEY (A, B) ON CONFLICT IGNORE)''')

这样，重复的行（也就是那些违反主键唯一性约束的行）会被悄悄跳过，而空值则会导致操作中断（这会引发一个sqlite3异常）。这一切都不需要提前过滤空值或处理sqlite3 API的错误信息。我们现在可以简单地调用con.executemany()，不需要再做其他处理：

with con:
    con.executemany('INSERT INTO ABCD VALUES (?, ?, ?, ?)', inputs)

Answer 3

我最后采取了这样的做法：

con = sqlite3.connect(':MEMORY:')
con.execute('''CREATE TABLE ABCD (A TEXT NOT NULL,
                                  B TEXT NOT NULL,
                                  C TEXT NOT NULL,
                                  D TEXT NOT NULL,
                                  PRIMARY KEY (A, B))''')
with con:
    for a, b, c, d in inputs:
        if any(elem is None for elem in (a, b, c, d)):
            raise ValueError('Fields must not be empty.')
        con.execute('INSERT OR IGNORE INTO ABCD VALUES (?, ?, ?, ?)',
                    (a, b, c, d))
con.close()

这样一来，在执行数据库操作之前，可以“手动”捕捉到空值。如果在执行execute时发生任何错误（比如违反了UNIQUE约束），那么这个输入就会被跳过。需要注意的是，INSERT OR IGNORE并不是说忽略唯一性约束，而是说跳过某一行输入。

这个解决方案的缺点是对空值的检查做了两次。不过我觉得这也还好，因为这个操作应该比较简单。相比于解析错误信息，这种方式更清晰，而且可能对未来的变化（比如pysqlite更新可能会改变错误信息的某些细节）更稳健。

致谢：这个想法是在与Lutz讨论中产生的，Martijn也独立提出了这个建议。