无法捕获SQLAlchemy的IntegrityError

Question

我尽力了，但似乎无法正确捕捉到sqlalchemy的完整性错误：

from sqlalchemy import exc

try:
    insert_record()
except exc.IntegrityError, exc:
    print exc # this is never called
    handle_elegantly() # this is never called

这就像大家预期的那样：

IntegrityError: (IntegrityError) insert or update on table "my_table" 
                violates foreign key constraint "my_table_some_column_fkey"

我尝试过明确地：

from sqlalchemy.exc import IntegrityError

更新：

我发现了一些似乎与这里发生的事情相符的内容，完整性错误不会在会话被刷新到数据库之前抛出，而是在try/except块执行之后才会出现：尝试捕捉sqlalchemy的完整性错误

然而，在try块中添加session.flush()会导致出现InvalidRequestError：

ERROR:root:This Session's transaction has been rolled back due to a previous 
           exception during flush. To begin a new transaction with this Session, 
           first issue Session.rollback(). 
           Original exception was: (IntegrityError)

Answer 1

SQLALCHEMY_COMMIT_ON_TEARDOWN = False

当然可以！请把你想要翻译的内容发给我，我会帮你用简单易懂的语言解释清楚。

Answer 2

这个回答和上面的回答是一样的，我只是想加一段额外的代码，如果你想在except中显示一些动态错误，可以把它作为JsonResponse返回，或者其他形式。

注意：输出这种类型的错误在安全性上并不好。

from sqlalchemy import exc


try:
     db.session.add(resource)
     return db.session.commit()
except exc.IntegrityError as e:
     err_msg = str(e.orig).split(':')[-1].replace('\n', '').strip()
     print(err_mg)
     return err_mg 

exc.IntegrityError有一个叫orig的键，里面包含了这个信息：

duplicate key value violates unique constraint "ix_table_column"
DETAIL:  Key (column)=(some_data) already exists.

这个信息的类型是<class 'driver.errors.UniqueViolation'>，而不是str，所以我们需要把它转换成str，然后通过一些简单的字符串处理，我们就能得到确切的错误信息，这里是：

Key (column)=(some_data) already exists.

Answer 3

在你捕捉到错误之后，记得要加上 session.rollback() 这一行代码：

try:
    session.flush()
except IntegrityError:
    session.rollback()

Answer 4

一旦出现了 IntegrityError 错误，不管你有没有处理这个错误，你正在使用的会话就会失效。正如第二条错误信息所说的，要想在这个会话中开始一个新事务，首先需要执行 Session.rollback()。 如果你想继续使用这个会话，就得执行 session.rollback()。

我不能确定，但我猜可能是你或者你的网络框架在尝试继续使用那个引发了 IntegrityError 的会话。建议你在捕获到这个异常后，或者在你的 handle_elegantly 函数中执行 session.rollback()。

如果你运行下面的代码，你就会明白我说的是什么意思：

from sqlalchemy import types
from sqlalchemy import exc
from sqlalchemy import create_engine
from sqlalchemy.schema import Column
from zope.sqlalchemy import ZopeTransactionExtension
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session

Base = declarative_base()


class User(Base):
    __tablename__ = 'user'
    name = Column(types.String, primary_key=True)


def handle_elegantly(name):
    session = DBSession()
    session.add(User(name=name))
    session.flush()
    print 'Exception elegantly handled!!\n'


def pretend_view(request):
    """Pretend view in a Pyramid application using pyramid_tm"""
    session = DBSession()
    user = User()
    print '\n-------Here we rollback before continuing -------'
    try:
        session.add(user)
        session.flush()
    except exc.IntegrityError:
        session.rollback()
        handle_elegantly('This will run fine')

    print '\n------- Here we do not, and this will error -------'
    try:
        session.add(user)
        session.flush()
    except exc.IntegrityError:
        handle_elegantly('Exception will be raised')


if __name__ == '__main__':
    engine = create_engine('sqlite://')
    global DBSession
    DBSession = scoped_session(
        sessionmaker(extension=ZopeTransactionExtension()))
    DBSession.configure(bind=engine)
    Base.metadata.bind = engine
    Base.metadata.create_all()
    pretend_view("dummy request")

Answer 5

我在我的Flask应用中也有同样的需求，我是这样处理的，效果很好：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import exc

db = SQLAlchemy(Flask(__name__))

try:
     db.session.add(resource)
     return db.session.commit()
except exc.IntegrityError:
     db.session.rollback()