无法捕获 SQLAlchemy IntegrityError

Posted 2023-02-23

【中文标题】无法捕获 SQLAlchemy IntegrityError

【英文标题】：cannot catch SQLAlchemy IntegrityError

【发布时间】：2014-08-22 17:21:27

【问题描述】：

尽我所能，我似乎无法正确捕捉到 sqlalchemy IntegrityError：

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/exceptblocks 之后：Trying to catch integrity error with 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)

【问题讨论】：

你是 100% 发生这种情况的吗？

您使用的是哪个数据库？

另外，将IntegrityError 更改为DatabaseError 有帮助吗？谢谢。

@alecxe DatabaseError 似乎没有帮助，我正在使用 Postgres。

@MartinKonecny 是的，我很确定。印刷声明比比皆是。我在这里缺少一些东西，但是什么？也许在会话中处理事务的方式中，在会话刷新到数据库之前不会抛出 IntegrityError，如下所述：***.com/questions/11313935/…

【参考方案1】：

我的 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()

【讨论】：

谢谢，我尝试使用 exc.IntegrityError 而不先从 sqlalchemy 导入它

【参考方案2】：

只要引发IntegrityError，无论您是否发现错误，您正在处理的会话都会失效。由于第二条错误消息指示您 To begin a new transaction with this Session, first issue Session.rollback(). 继续使用会话，您需要发出 session.rollback()

我不能肯定地说，但我猜您或您的 Web 框架正在尝试继续使用以某种方式引发 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")

【参考方案3】：

发现错误后必须添加session.rollback()：

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

【参考方案4】：

SQLALCHEMY_COMMIT_ON_TEARDOWN = False