Невозможно поймать SQLAlchemy IntegrityError при удалении

Question

Я знаю, что есть много других вопросов о той же самой проблеме, но я пробовал их ответы, и никто не работал до сих пор.

Я пытаюсь удалить записи из таблицы, которая имеет отношения с другими таблицами. Внешний ключ в этих таблицах: nullable=false, поэтому попытка удалить запись, которая используется другой таблицей, должна вызвать исключение.

Но даже при общении с инструкцией delete с помощью catchall try-except ошибка по-прежнему не обнаружена, поэтому я подозреваю, что исключение может быть поднято в другом месте.

Я использую SQLite с SQLAlchemy в структуре Pyramid, и мой сеанс настроен с помощью ZopeTransactionExtension.

Это, как я пытаюсь удалить: В 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() 

В models.py я создал DBSession и мои модели:

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

И в __init__.py Я настраиваю сеанс следующим образом:

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 

Используя эту настройку, я получаю

IntegrityError: (IntegrityError) NOT NULL constraint failed

Traceback here.

Если я заменяю transaction.commit() с DBSession.flush(), я получаю

ResourceClosedError: This transaction is closed

И если я удалить transaction.commit(), я все еще получаю ту же ошибку, но без четкой точки происхождения.

ОБНОВЛЕНИЕ: Я провел некоторые тесты носа, и в некоторых случаях, но не во всех случаях, исключение обрабатывалось правильно.

В моих тестах я импортировать сессию и настроить его:

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) 

Кто-нибудь знает, что происходит?

Answer 1

Возможно, вы просто «делаете это неправильно». Ваш вопрос затрагивает два вопроса. Обработка ошибок уровня транзакций, возникающих из-за ошибок целостности базы данных и моделирования кода/моделей приложений/запросов для реализации бизнес-логики. Мой ответ сосредоточен на написании кода, который соответствует общим шаблонам при использовании pyramid_tm для управления транзакциями и sqlalchemy как ORM.

В Pyramid, если вы настроили свой сеанс (который делает для вас файл автоматически), чтобы использовать ZopeTransactionExtension, тогда сеанс не очищается и не будет выполняться до тех пор, пока не будет выполнено представление. Если вы хотите поймать какие-либо ошибки SQL самостоятельно в своем представлении, вам нужно заставить флеш отправить SQL на двигатель. DBSession.flush() должен сделать это после удаления (...).

Если вы поднимете код возврата из 4xx/5xx как исключение пирамиды HTTPConflict, транзакция будет прервана.

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

Это excerpt from todopyramid/models.py основных моментов, как удалить элемент коллекции без использования DBSession объекта.

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

Это sample code from pyramid_blogr ясно показывают, как простой пирамидой код вид для удаления элементов базы данных SQL может выглядеть следующим образом. Обычно вам не нужно взаимодействовать с транзакцией. Это функция - как рекламируется как one the unique feature of pyramid. Просто выберите любой из доступных учебных пособий по пирамиде, которые используют sqlalchemy и стараются как можно больше придерживаться шаблонов. Если вы решите проблему на уровне модели приложения, механизм транзакций будет скрываться в фоновом режиме, если у вас нет четкой потребности в его услугах.

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

Чтобы обеспечить осмысленные сообщения об ошибках пользователей приложения вы либо отлавливать ошибки на контрсилах базы данных в модели базы данных слоя или слой представления пирамиды. Ловля SQLAlchemy исключения для обеспечения сообщения об ошибках может выглядеть в этом 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) 

Ссылки

• Trying to catch integrity error with SQLAlchemy
• pyramid_tm Usage
• What the Zope Transaction Manager Means To Me (and you)

Answer 2

Не уверен, что если это помогает - я сделал не совсем захватывают от t он трассирует то, что пойдет не так, потребуется больше времени. Но вы можете использовать диспетчер транзакций следующим образом:

from sqlalchemy.exc import IntegrityError 


try: 
    with transaction.manager: 
     deletethis = DBSession.query(foo).filter_by(id=deleteid).first() 
     qry = DBSession.delete(deletethis) 
     if qry == 0: 
      return HTTPNotFound() 
    # transaction.manager commits when with context manager exits here 
except IntegrityError: 
    DBSession.rollback() 
    return HTTPConflict() 

return HTTPOk()