no puede capturar SQLAlchemy IntegrityError

Por más que lo intente, parece que no puedo capturar correctamente el IntegrityError sqlalchemy:

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

Como lo que uno podría esperar:

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

He intentado explícitamente:

 from sqlalchemy.exc import IntegrityError 

ACTUALIZAR:

Encontré algo que parece encajar en lo que está sucediendo aquí, donde el error de integridad no se lanza hasta que la sesión se vacía en la base de datos, y después de que se ejecutaron los bloques de try / except : tratando de detectar el error de integridad con SQLAlchemy

Sin embargo, al agregar session.flush() en el bloque try se obtiene un 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) 

Tan pronto como se IntegrityError , independientemente de si ha detectado o no el error, la sesión en la que estaba trabajando se invalida. Como el segundo mensaje de error le indica: To begin a new transaction with this Session, first issue Session.rollback(). , para continuar usando la sesión necesitarás emitir una session.rollback()

No puedo decirlo con seguridad, pero supongo que usted o su marco de trabajo web intentan continuar utilizando la sesión que provocó el IntegrityError de alguna manera. Le recomiendo que emita un session.rollback() después de detectar la excepción o en su función handle_elegantly .

Si ejecuta el siguiente verás lo que quiero decir:

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

Tengo la misma necesidad en mi aplicación Flask, la manejo como se muestra a continuación y funciona:

 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 as e: db.session().rollback() 

SQLALCHEMY_COMMIT_ON_TEARDOWN = Falso