Unable to catch SQLAlchemy IntegrityError when deleting

Question

I'm aware there are many other questions about the exact same issue, but I've tried their answers and none have worked so far.

I'm trying to delete records from a table which has relationships with other tables. The foreign key in those tables are nullable=false, so trying to delete a record which is in use by another table should raise an exception.

But even when surrounding the delete statement with a catchall try-except the error is still not caught, so I suspect the exception may be raised somewhere else.

I'm using SQLite with SQLAlchemy in a Pyramid framework, and my session is configured with the ZopeTransactionExtension.

This is how I'm trying to delete: In 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()

In models.py I set up DBSession and my models:

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')

And in __init__.py I configure my session like so:

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

Using this setup I get

IntegrityError: (IntegrityError) NOT NULL constraint failed

Traceback here.

If I replace transaction.commit() with DBSession.flush(), I get

ResourceClosedError: This transaction is closed

And if I remove the transaction.commit(), I still get the same error, but without a clear point of origin.

UPDATE: I ran some nose tests, and in some cases, but not all, the exception was handled correctly.

In my tests I import the session and configure it:

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)

Does anyone know what's going on?

Sascha Gottfried · Accepted Answer

Might be you just "doing it wrong". Your question addresses two issues. Handling transaction level errors raised by database integrity errors and modelling application code/models/queries to implement business logic. My answer focuses on writing code that fits common patterns while using pyramid_tm for transaction management and sqlalchemy as an ORM.

In Pyramid, if you've configured your session (which the scaffold does for you automatically) to use the ZopeTransactionExtension, then session is not flushed/committed until after the view has executed. If you want to catch any SQL errors yourself in your view, you need to force a flush to send the SQL to the engine. DBSession.flush() should do it after the delete(...).

If you raise any of 4xx/5xx HTTP return codes like pyramid exception HTTPConflict the transaction will be aborted.

@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()

This excerpt from todopyramid/models.py highlights how to delete a collection item without using DBSession object.

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

This 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. Just pick any of the available pyramid tutorials that use sqlalchemy and try to stick to the patterns as much as possible. If you address the problem at the application model level the transaction machinery will hide in the background unless you have a clear need for its services.

@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'))

To provide meaningful error messages to application users you either catch errors on database contraints at database model layer or at pyramid view layer. Catching sqlalchemy exceptions to provide error messages could look like in this 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)

References

Unable to catch SQLAlchemy IntegrityError when deleting

Answers (2)

Related Questions