Попытка поймать ошибку целостности с помощью SQLAlchemy

Ответ 1

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

Обновление

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

def create_unique_object(db, max_attempts=3):
    while True:
        sp = transaction.savepoint()
        try:
            obj = MyObject()
            obj.identifier = uuid.uuid4().hex
            db.add(obj)
            db.flush()
        except IntegrityError:
            sp.rollback()
            max_attempts -= 1
            if max_attempts < 1:
                raise
        else:
            return obj

obj = create_unique_object(DBSession)

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

Ответ 2

Что вам нужно сделать, это поймать общее исключение и вывести его класс; то вы можете сделать исключение более конкретным.

except Exception as ex:
    print ex.__class__

Ответ 3

Не может быть никаких операций с базой данных до DBSession.commit(), поэтому IntegrityError возникает позже в стеке после того, как код контроллера, который имеет try/except, уже вернулся.

Ответ 4

Изменить: отредактированный ответ выше - лучший способ сделать это, используя откат.

-

Если вы хотите обрабатывать транзакции в середине приложения пирамиды или что-то, где автоматическая фиксация транзакции выполняется в конце последовательности, нет никакой магии, которая должна произойти.

Не забудьте запустить новую транзакцию, если предыдущая транзакция завершилась с ошибкой.

Вот так:

def my_view(request):
   ... # Do things
   if success:
     try:
       instance = self._instance(**data)
       DBSession.add(instance)
       transaction.commit()
       return {'success': True}
     except IntegrityError as e:  # <--- Oh no! Duplicate unique key
       transaction.abort()
       transaction.begin() # <--- Start new transaction
       return {'success': False}

Обратите внимание, что вызов .commit() в успешной транзакции прекрасен, поэтому нет необходимости запускать новую транзакцию после успешного вызова.

Вам нужно только прервать транзакцию и запустить новую, если транзакция находится в состоянии отказа.

(Если транзакция не была такой poc, вы могли бы использовать точку сохранения и откат назад к точке сохранения вместо того, чтобы начинать новую транзакцию; к сожалению, это невозможно, так как попытка фиксации делает недействительным известную предыдущую точку сохранения. да?) (edit: < --- Оказывается, я ошибаюсь в этом...)

Ответ 5

Вот как я это делаю.

from contextlib import(
        contextmanager,
        )


@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()



def create_user(email, firstname, lastname, password):

    new_user = Users(email, firstname, lastname, password)

    try:

        with session_scope() as session:

            session.add(new_user)

    except sqlalchemy.exc.IntegrityError as e:
        pass

http://docs.sqlalchemy.org/en/latest/orm/session_basics.html#when-do-i-construct-a-session-when-do-i-commit-it-and-when-do-i-close-it