删除时无法捕获 SQLAlchemy IntegrityError
Unable to catch SQLAlchemy IntegrityError when deleting
我知道还有很多关于完全相同问题的其他问题，但我已经尝试了他们的答案并且 none 到目前为止已经奏效了。
我正在尝试从与其他 table 有关系的 table 中删除记录。 table 中的外键是 nullable=false，因此尝试删除另一个 table 正在使用的记录应该引发异常。
但是即使在 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 constraint failed
追溯here.
如果我将 transaction.commit() 替换为 DBSession.flush()，我会得到
ResourceClosedError: This transaction is closed
如果我删除 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)
有人知道这是怎么回事吗？
也许你就是 "doing it wrong"。你的问题涉及两个问题。处理由数据库完整性错误和建模应用程序 code/models/queries 引发的事务级错误以实现业务逻辑。我的回答侧重于编写符合常见模式的代码，同时使用 pyramid_tm 进行事务管理和将 sqlalchemy 作为 ORM。
在 Pyramid 中，如果您已将会话（脚手架自动为您完成）配置为使用 ZopeTransactionExtension，那么在视图执行之前会话不会 flushed/committed。 如果您想自己在视图中捕获任何 SQL 错误，您需要强制刷新以将 SQL 发送到引擎 。 DBSession.flush() 应该在 delete(...) 之后执行。
如果您提出任何 4xx/5xx HTTP return 代码，例如金字塔异常 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
"""
todo_item = self.todo_list.filter(
TodoItem.id == todo_id)
todo_item.delete()
这个sample code from pyramid_blogr show clearly how simple pyramid view code to delete SQL database items could look like. Usually you do not have to interact with the transaction. This is a feature - as advertised as 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)
不确定这是否有帮助 - 我没有完全从回溯中捕捉到哪里出了问题，需要更多时间。但是你可以像这样使用事务管理器：
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()
我知道还有很多关于完全相同问题的其他问题，但我已经尝试了他们的答案并且 none 到目前为止已经奏效了。
我正在尝试从与其他 table 有关系的 table 中删除记录。 table 中的外键是 nullable=false，因此尝试删除另一个 table 正在使用的记录应该引发异常。
但是即使在 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 constraint failed
追溯here.
如果我将 transaction.commit() 替换为 DBSession.flush()，我会得到
ResourceClosedError: This transaction is closed
如果我删除 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)
有人知道这是怎么回事吗？
也许你就是 "doing it wrong"。你的问题涉及两个问题。处理由数据库完整性错误和建模应用程序 code/models/queries 引发的事务级错误以实现业务逻辑。我的回答侧重于编写符合常见模式的代码，同时使用 pyramid_tm 进行事务管理和将 sqlalchemy 作为 ORM。
在 Pyramid 中，如果您已将会话（脚手架自动为您完成）配置为使用 ZopeTransactionExtension，那么在视图执行之前会话不会 flushed/committed。 如果您想自己在视图中捕获任何 SQL 错误，您需要强制刷新以将 SQL 发送到引擎 。 DBSession.flush() 应该在 delete(...) 之后执行。
如果您提出任何 4xx/5xx HTTP return 代码，例如金字塔异常 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
"""
todo_item = self.todo_list.filter(
TodoItem.id == todo_id)
todo_item.delete()
这个sample code from pyramid_blogr show clearly how simple pyramid view code to delete SQL database items could look like. Usually you do not have to interact with the transaction. This is a feature - as advertised as 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)
不确定这是否有帮助 - 我没有完全从回溯中捕捉到哪里出了问题，需要更多时间。但是你可以像这样使用事务管理器：
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()