删除时无法捕获 SQLAlchemy IntegrityError

Question

我知道关于完全相同的问题还有许多其他问题，但我已经尝试了他们的答案，但到目前为止没有一个有效。

我正在尝试从与其他表有关系的表中删除记录。这些表中的外键是nullable=false，因此尝试删除另一个表正在使用的记录应该会引发异常。

但是，即使在 delete 语句周围加上一个包罗万象的 try-except，错误仍然没有被捕获，所以我怀疑该异常可能会在其他地方引发。

我在 Pyramid 框架中使用 SQLite 和 SQLAlchemy，我的会话配置为 ZopeTransactionExtension。

这就是我尝试删除的方式： 在views.py中

from sqlalchemy.exc import IntegrityError
from project.app.models import (
    DBSession,
    foo)

@view_config(route_name='fooview', renderer='json', permission='view')
def fooview(request):
    """ The fooview handles different cases for foo
        depending on the http method
    """
    if request.method == 'DELETE':
        if not request.has_permission('edit'):
            return HTTPForbidden()

        deleteid = request.matchdict['id']
        deletethis = DBSession.query(foo).filter_by(id=deleteid).first()

        try:
           qry = DBSession.delete(deletethis)
           transaction.commit()
           if qry == 0:
               return HTTPNotFound(text=u'Foo not found')
       except IntegrityError:
           DBSession.rollback()
           return HTTPConflict(text=u'Foo in use')

        return HTTPOk()

在 models.py 中，我设置了 DBSession 和我的模型：

from zope.sqlalchemy import ZopeTransactionExtension
from sqlalchemy.orm import (
    scoped_session,
    sessionmaker,
    relationship,
    backref,
)

DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension('changed')))
Base = declarative_base()

class foo(Base):
    """ foo defines a unit used by bar
    """
    __tablename__ = 'foo'
    id = Column(Integer, primary_key=True)
    name = Column(Text(50))

    bars = relationship('bar')

class bar(Base):
    __tablename__ = 'bar'
    id = Column(Integer, primary_key=True)
    fooId = Column(Integer, ForeignKey('foo.id'), nullable=False)

    foo = relationship('foo')

在 __init__.py 中，我像这样配置我的会话：

from project.app.models import (
    DBSession,
    Base,
)

def main(global_config, **settings):
    """ This function returns a Pyramid WSGI application.
    """
    engine = engine_from_config(settings, 'sqlalchemy.')
    # fix for association_table cascade delete issues
    engine.dialect.supports_sane_rowcount = engine.dialect.supports_sane_multi_rowcount = False
    DBSession.configure(bind=engine)
    Base.metadata.bind = engine

使用这个设置我得到了

IntegrityError: (IntegrityError) NOT NULL 约束失败

回溯here。

如果我用DBSession.flush() 替换transaction.commit()，我会得到

ResourceClosedError: 此事务已关闭

如果我删除transaction.commit()，我仍然会收到同样的错误，但没有明确的原点。

更新： 我进行了一些鼻子测试，在某些情况下（但不是全部），异常得到了正确处理。

在我的测试中，我导入会话并对其进行配置：

from optimate.app.models import (
    DBSession,
    Base,
    foo)

def _initTestingDB():
    """ Build a database with default data
    """
    engine = create_engine('sqlite://')
    Base.metadata.create_all(engine)
    DBSession.configure(bind=engine)
    with transaction.manager:
        # add test data

class TestFoo(unittest.TestCase):
    def setUp(self):
        self.config = testing.setUp()
        self.session = _initTestingDB()

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def _callFUT(self, request):
        from project.app.views import fooview
        return fooview(request)

    def test_delete_foo_keep(self):
        request = testing.DummyRequest()
        request.method = 'DELETE'
        request.matchdict['id'] = 1
        response = self._callFUT(request)
        # foo is used so it is not deleted
        self.assertEqual(response.code, 409)

    def test_delete_foo_remove(self):
        _registerRoutes(self.config)
        request = testing.DummyRequest()
        request.method = 'DELETE'
        request.matchdict['id'] = 2
        response = self._callFUT(request)
        # foo is not used so it is deleted
        self.assertEqual(response.code, 200)

有人知道怎么回事吗？

Answer 1

可能只是你“做错了”。你的问题解决了两个问题。处理由数据库完整性错误引发的事务级错误，并对应用程序代码/模型/查询进行建模以实现业务逻辑。我的回答侧重于编写符合常见模式的代码，同时使用 pyramid_tm 进行事务管理和将 sqlalchemy 作为 ORM。

在 Pyramid 中，如果您已将会话（脚手架自动为您执行）配置为使用 ZopeTransactionExtension，则在视图执行之前不会刷新/提交会话。 如果您想自己在视图中捕获任何 SQL 错误，则需要强制刷新以将 SQL 发送到引擎。 DBSession.flush() 应该在 delete(...) 之后执行。

如果您提出任何 4xx/5xx HTTP 返回码，例如金字塔异常 HTTPConflict，事务将被中止。

@view_config(route_name='fooview', renderer='json', permission='view')
def fooview(request):
    """ The fooview handles different cases for foo
        depending on the http method
    """
    if request.method == 'DELETE':
        if not request.has_permission('edit'):
            return HTTPForbidden()

        deleteid = request.matchdict['id']
        deletethis = DBSession.query(foo).filter_by(id=deleteid).first()
        if not deletethis:
            raise HTTPNotFound()

        try:
           DBSession.delete(deletethis)
           DBSession.flush()
        except IntegrityError as e:
           log.debug("delete operation not possible for id {0}".format(deleteid)
           raise HTTPConflict(text=u'Foo in use')

        return HTTPOk()

此excerpt from todopyramid/models.py 重点介绍了如何在不使用DBSession 对象的情况下删除集合项。

def delete_todo(self, todo_id):
    """given a todo ID we delete it is contained in user todos 

    delete from a collection
    http://docs.sqlalchemy.org/en/latest/orm/session.html#deleting-from-collections
    https://stackoverflow.com/questions/10378468/deleting-an-object-from-collection-in-sqlalchemy"""
    todo_item = self.todo_list.filter(
            TodoItem.id == todo_id)

    todo_item.delete()

sample code from pyramid_blogr 清楚地展示了删除 SQL 数据库项目的简单金字塔视图代码的外观。通常您不必与事务交互。这是一项功能 - 宣传为 one the unique feature of pyramid。只需选择任何使用 sqlalchemy 的可用金字塔教程并尝试尽可能地坚持模式。如果您在应用程序模型级别解决问题，除非您明确需要其服务，否则事务机制将隐藏在后台。

@view_config(route_name='blog_action', match_param="action=delete", permission='delete')
def blog_delete(request):
    entry_id = request.params.get('id', -1)
    entry = Entry.by_id(entry_id)
    if not entry:
        return HTTPNotFound()
    DBSession.delete(entry)
    return HTTPFound(location=request.route_url('home'))

要向应用程序用户提供有意义的错误消息，您可以在数据库模型层或金字塔视图层捕获数据库约束中的错误。捕获 sqlalchemy 异常以提供错误消息可能类似于 sample code

from sqlalchemy.exc import OperationalError as SqlAlchemyOperationalError

@view_config(context=SqlAlchemyOperationalError)
def failed_sqlalchemy(exception, request):
    """catch missing database, logout and redirect to homepage, add flash message with error

    implementation inspired by pylons group message 
    https://groups.google.com/d/msg/pylons-discuss/BUtbPrXizP4/0JhqB2MuoL4J
    """
    msg = 'There was an error connecting to database'
    request.session.flash(msg, queue='error')
    headers = forget(request)

    # Send the user back home, everything else is protected  
    return HTTPFound(request.route_url('home'), headers=headers)

参考文献

• Trying to catch integrity error with SQLAlchemy
• pyramid_tm Usage
• What the Zope Transaction Manager Means To Me (and you)

Answer 2

不确定这是否有帮助 - 我没有从追溯中完全捕捉到哪里出了问题，需要更多时间。但是你可以像这样使用事务管理器：

from sqlalchemy.exc import IntegrityError


try:
   with transaction.manager:
        deletethis = DBSession.query(foo).filter_by(id=deleteid).first()
        qry = DBSession.delete(deletethis)
        if qry == 0:
            return HTTPNotFound()
    # transaction.manager commits when with context manager exits here
except IntegrityError:
   DBSession.rollback()
   return HTTPConflict()

return HTTPOk()