не может поймать SQLAlchemy IntegrityError

Question

Try как я мог, я не могу показаться, чтобы правильно уловить SQLAlchemy IntegrityError:

from sqlalchemy import exc 

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

Как то, что можно было бы ожидать:

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

Я попытался явно :

from sqlalchemy.exc import IntegrityError 

UPDATE:

я нашел то, что, кажется, подходит то, что здесь, где целостность ошибки не выдается, пока сеанс не смывается в БД происходит, и после того, как try/except блоки были выполнены: Trying to catch integrity error with SQLAlchemy

Однако добавление session.flush() в try блоке дает 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

Как только IntegrityError поднимается, независимо от того, был ли или не вы поймали ошибку, сеанс вы работали в аннулируется. Поскольку сообщение второй ошибки инструктирует вас, To begin a new transaction with this Session, first issue Session.rollback()., чтобы продолжить использование сеанса вам нужно выдать session.rollback()

Я не могу сказать точно, но я предполагаю, что вы или ваш веб-фреймворк пытается продолжать использовать сессию который каким-то образом поднял IntegrityError. Я рекомендую вам выдать session.rollback() либо после того, как вы поймаете исключение, либо в своей функции handle_elegantly.

Если вы запустите ниже вы увидите, что я имею в виду:

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 2

У меня такая же потребность в моем Колба приложения, я справиться с этим, как показано ниже, и она работает:

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