IntegrityError в pysqlite: отличить "NOT NULL" от нарушения "UNIQUE"

Question

В писклите, нарушая 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

Это то, что я в итоге сделал:

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, которое может изменить некоторые детали в сообщении об ошибке).

Кредиты: идея возникла из обсуждения с Лутцем. Это было также независимо предложено Мартийном.

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 исключение). Все это достигается без предварительной фильтрации данных по значениям Null/None или с сообщениями об ошибках sqlite3 API. Теперь мы можем просто позвонить con.executemany(), без дальнейших церемоний:

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

Answer 3

Ниже приведен рабочий код:

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
>>> 

Надежды, могут помочь вам.