python - no puede capturar SQLAlchemy IntegrityError

exception exception-handling

Question

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 produce hasta que la sesión se vacía en la base de datos, y después de que se ejecutan 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)

Answer 1

SQLALCHEMY_COMMIT_ON_TEARDOWN = Falso

Answer 2

Tan pronto como se IntegrityError , independientemente de si ha detectado el error o no, 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 certeza, pero supongo que usted o su marco de trabajo web intentan continuar utilizando la sesión que provocó 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")

Answer 3

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