cannot catch SQLAlchemy IntegrityError

Question

Try as I might, I can't seem to catch the sqlalchemy IntegrityError correctly:

from sqlalchemy import exc

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

As what one might expect:

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

I've tried to explicitly:

from sqlalchemy.exc import IntegrityError

UPDATE:

I found something that seems to fit what's happening here, where Integrity Error isn't thrown until the session is flushed to the db, and after the try/exceptblocks have been executed: Trying to catch integrity error with SQLAlchemy

However, adding session.flush() in the try block yields an 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)

Solution 1

I have the same need in my Flask application, I handle it like below and it works:

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

Solution 2

As soon as the IntegrityError is raised, regardless of whether or not you've caught the error, the session you were working in is invalidated. As the second error message is instructing you, To begin a new transaction with this Session, first issue Session.rollback()., to continue using the session you'll need to issue a session.rollback()

I cannot say for sure, but I am guessing you or your web framework is attempting to continue using the session which raised the IntegrityError in some way. I recommend you issue a session.rollback() either after you catch the exception or in your handle_elegantly function.

If you run the below you'll see what I mean:

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

Solution 3

You have to add session.rollback() after you catch an error:

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