logo标签列表

如何区分SQLAlchemy的完整性错误(IntegrityError)的原因？

sqlalchemy
5
SQLAlchemy在事务出现数据完整性问题时似乎只会抛出一个通用的IntegrityError异常。当然，确切的查询和错误消息包含在异常中，这对于人类调试程序是足够的。然而，当为异常编写错误处理代码时，似乎没有好的方法可以检查哪个表上的哪个约束条件导致了错误。此外，异常是由session.commit()行引发的，而不是实际产生错误的行，因此我也无法使用多个try/except块进行区分。
除了尝试以编程方式解析错误消息和/或查询之外，是否有一种方法可以区分重复主键错误、外键错误或失败的CHECK约束等？或者仅仅有一种方法可以告诉我哪个表的哪个列违反了数据完整性？或者只要等待事务提交就立即引发异常的方法？
4个回答
4
IntegrityError实例具有orig和statement属性，可以检查以获取错误消息和失败的SQL语句。

给定这个模型：

class Foo(Base):
    __tablename__ = 'foo20201209'

    id = sa.Column(sa.Integer, primary_key=True)
    bar = sa.Column(sa.String(2), unique=True)
    baz = sa.Column(sa.Integer, sa.CheckConstraint('baz >= 0'), default=0)

这段代码：
conn_strings = ['postgresql+psycopg2:///test',
                'mysql+mysqlconnector:///test',
                'sqlite://']


for cs in conn_strings:
    engine = sa.create_engine(cs)
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    session = orm.Session(bind=engine)

    for kwds in [{'bar': 'a'}, {'bar': 'a'}, {'bar': 'b', 'baz': -11}]:
        session.add(Foo(**kwds))
        try:
            session.commit()
        except sa.exc.IntegrityError as ex:
            print(ex.orig)
            print(ex.statement)
            print()
            session.rollback()
    session.close()
    engine.dispose()

会产生以下输出：
duplicate key value violates unique constraint "foo20201209_bar_key"
DETAIL:  Key (bar)=(a) already exists.

INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s) RETURNING foo20201209.id

new row for relation "foo20201209" violates check constraint "foo20201209_baz_check"
DETAIL:  Failing row contains (3, b, -11).

INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s) RETURNING foo20201209.id

1062 (23000): Duplicate entry 'a' for key 'bar'
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s)

4025 (23000): CONSTRAINT `foo20201209.baz` failed for `test`.`foo20201209`
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s)

UNIQUE constraint failed: foo20201209.bar
INSERT INTO foo20201209 (bar, baz) VALUES (?, ?)

CHECK constraint failed: foo20201209
INSERT INTO foo20201209 (bar, baz) VALUES (?, ?)
1
感谢您的回复。问题在于错误消息和 SQL 语句的文本可能会因引擎而异，同样也可能会因版本而异。我可以阅读这些消息并知道查询失败的原因，但让程序解析它更难。理想情况下，我希望最终得到对 Foo 的引用，对 Foo.bar 的引用以及一个常量或枚举值，指示它是 CHECK、UNIQUE 还是任何其他类型的约束。看起来 SQLAlchemy 没有这种能力... - DMJ
2

如下DMJ所评论的我的答案，不同的数据库引擎在完整性错误发生时会发出不同的错误消息，而SQLAlchemy并不尝试以一致的方式呈现这些消息，因此问题的简短答案是：您无法从异常中获得有用信息，除非解析错误消息。

那么，如果约束以一致的方式命名并且这些名称出现在错误消息中，则解析错误消息可能不是那么困难。在这方面，SQLAlchemy提供了一些帮助：我们可以创建一个元数据对象，并定义一个约束命名规则（文档）。有了这个约定，我们创建的任何约束都将遵循它，我们可以匹配错误消息中的约束名称，并使用这些匹配项在元数据中查找表和约束对象。
以下是使用这些约定解析错误消息的示例。我没有涵盖所有可能的约束类型，也没有处理Sqlite的唯一键冲突消息，这些留给读者作为练习；-)
import re
import sqlalchemy as sa

convention = {
  "ix": 'ix_%(column_0_label)s',
  "uq": "uq_%(table_name)s_%(column_0_name)s",
  "ck": "ck_%(table_name)s_%(constraint_name)s",
  "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
  "pk": "pk_%(table_name)s"
}

metadata = sa.MetaData(naming_convention=convention)

tbl = sa.Table(
    't65189213',
    metadata,
    sa.Column('id', sa.Integer, primary_key=True),
    sa.Column('foo', sa.String(2), unique=True),
    sa.Column('bar', sa.Integer, default=0),
    sa.CheckConstraint('bar >= 0', name='positive_bar'),
)

# Basic pattern to match constraint names (not exhaustive).
pattern = r'(?P<type_key>uq|ck)_(?P<table>[a-z][a-z0-9]+)_[a-z_]+'
regex = re.compile(pattern)


def parse_exception(ex):
    # Returns an informative message, or the original error message.
    # We could return the table, constraint object etc. instead.
    types = {'ck': 'Check constraint', 'uq': 'Unique key'}
    m = regex.search(str(ex.orig))
    if m:
        type_ = types[m.groupdict()['type_key']]
        table_name = m.groupdict()['table']
        constraint_name = m.group(0)
        table = metadata.tables[table_name]
        constraint = next(c for c in table.constraints if c.name == constraint_name)
        columns = ','.join(constraint.columns.keys())
        return f'{type_} {constraint_name} has been violated on table {table_name!r} columns: {columns}'
    return f'{ex.orig}'


conn_strings = ['postgresql+psycopg2:///test',
                'mysql+mysqlconnector://root:root@localhost/test',
                'sqlite://']


for cs in conn_strings:
    engine = sa.create_engine(cs, future=True)
    tbl.drop(engine, checkfirst=True)
    tbl.create(engine)

    with engine.connect() as conn:
        print(engine.dialect.name)
        for kwds in [{'foo': 'a'}, {'foo': 'a'}, {'foo': 'b', 'bar': -11}]:
            try:
                conn.execute(tbl.insert(), kwds)
                conn.commit()
            except sa.exc.IntegrityError as ex:
                print(parse_exception(ex))
                conn.rollback()
    print('-'*10)
    engine.dispose()

输出：

postgresql
Unique key uq_t65189213_foo has been violated on table 't65189213' columns: foo
Check constraint ck_t65189213_positive_bar has been violated on table 't65189213' columns: 
----------
mysql
Unique key uq_t65189213_foo has been violated on table 't65189213' columns: foo
Check constraint ck_t65189213_positive_bar has been violated on table 't65189213' columns: 
----------
sqlite
UNIQUE constraint failed: t65189213.foo
Check constraint ck_t65189213_positive_bar has been violated on table 't65189213' columns: 
----------
1

我也遇到了同样的问题，对于约束条件来说，更好的解决方案是给它们命名，然后解析异常信息。无论如何，在我的SQLAlchemy==1.4.28版本中，唯一约束条件名称仍未在抛出的异常中呈现。

考虑下面的例子：

class M(Model):
        a = Column(String)
        i = Column(Integer)
    
UniqueConstraint(M.i, name="my unique")
CheckConstraint(0 <= M.i, name="my check")

def test_test():
    db = create_test_database(Model)
    try:
        with db.session() as s:
        
            # Here we break our UNIQUE constraint
            try:
                with s.begin_nested():
                    s.add_all(
                        [
                            M(a="Alice", i=1),
                            M(a="Bob", i=1),
                        ]
                    )
            except IntegrityError as err:
                rich.inspect(err)

            # Here we break our CHECK constraint
            s.add(M(a="Alice", i=-1))

    except IntegrityError as err:
        rich.inspect(err)

这里是结果:
    ┌────────────────────── <class 'sqlalchemy.exc.IntegrityError'> ───────────────────────┐
    │ Wraps a DB-API IntegrityError.                                                       │
    │                                                                                      │
    │ ┌──────────────────────────────────────────────────────────────────────────────────┐ │
    │ │ IntegrityError('(sqlite3.IntegrityError) UNIQUE constraint failed: M.i')         │ │
    │ └──────────────────────────────────────────────────────────────────────────────────┘ │
    │                                                                                      │
    │                   args = ('(sqlite3.IntegrityError) UNIQUE constraint failed: M.i',) │
    │                   code = 'gkpj'                                                      │
    │ connection_invalidated = False                                                       │
    │                 detail = []                                                          │
    │        hide_parameters = False                                                       │
    │                ismulti = False                                                       │
    │                   orig = IntegrityError('UNIQUE constraint failed: M.i')             │
    │                 params = ('Bob', 1)                                                  │
    │              statement = 'INSERT INTO "M" (a, i) VALUES (?, ?)'                      │
    └──────────────────────────────────────────────────────────────────────────────────────┘
    ┌──────────────────────── <class 'sqlalchemy.exc.IntegrityError'> ─────────────────────────┐
    │ Wraps a DB-API IntegrityError.                                                           │
    │                                                                                          │
    │ ┌──────────────────────────────────────────────────────────────────────────────────────┐ │
    │ │ IntegrityError('(sqlite3.IntegrityError) CHECK constraint failed: my check')         │ │
    │ └──────────────────────────────────────────────────────────────────────────────────────┘ │
    │                                                                                          │
    │                   args = ('(sqlite3.IntegrityError) CHECK constraint failed: my check',) │
    │                   code = 'gkpj'                                                          │
    │ connection_invalidated = False                                                           │
    │                 detail = []                                                              │
    │        hide_parameters = False                                                           │
    │                ismulti = False                                                           │
    │                   orig = IntegrityError('CHECK constraint failed: my check')             │
    │                 params = ('Alice', -1)                                                   │
    │              statement = 'INSERT INTO "M" (a, i) VALUES (?, ?)'                          │
    └──────────────────────────────────────────────────────────────────────────────────────────┘

所以，看起来CHECK约束名总是会出现在异常字符串中的某个位置，你可以编写相应的代码。显然，这是关于UNIQUE约束的额外信息，因为你已经有了单词UNIQUE和字段名（在我的示例中是A.i）。我认为他们永远不会改变这些字符串的格式，但询问一下也是很有趣的。 警告：
问题在于我的代码涉及SQLite，而你的代码可能涉及另一个数据库，这些消息将有所不同，因为它们最初来自底层的数据库引擎，而不是SQLAlchemy本身。因此，你需要注意为这些字符串抽象出你的代码。
使用约束名称进行解析是一个好主意，因为它们经常出现在错误消息中。我添加了另一个带有一些演示代码的答案。 - snakecharmerb
1
我最终使用了session.flush()来更早地触发异常。 我在问题所在的行（这样我就可以百分之百确定异常不是由前面的行引起的）之前调用它一次，然后再在try / catch块中再次调用以查看是否有错误发生。
我承认我对这个解决方案并不完全满意，但我还没有找到其他解决方案。 如果有一个更好的解决方案，最好是能告诉我哪个表的哪个约束条件导致了错误，我仍然很想听听。 但是，这是一个可能会帮助某些人的解决方法。
网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接