尝试使用 SQLAlchemy 捕获完整性错误 [英] Trying to catch integrity error with SQLAlchemy

问题描述

我在尝试捕获错误时遇到问题.我正在使用 Pyramid/SQLAlchemy 并使用电子邮件作为主键制作了一个注册表单.问题是，当输入重复的电子邮件时，它会引发 IntegrityError，所以我试图捕捉该错误并提供一条消息，但无论我做什么我都无法捕捉到它，错误不断出现.

I'm having problems with trying to catch an error. I'm using Pyramid/SQLAlchemy and made a sign up form with email as the primary key. The problem is when a duplicate email is entered it raises a IntegrityError, so I'm trying to catch that error and provide a message but no matter what I do I can't catch it, the error keeps appearing.

try:
    new_user = Users(email, firstname, lastname, password)
    DBSession.add(new_user)
    return HTTPFound(location = request.route_url('new'))
except IntegrityError:
    message1 = "Yikes! Your email already exists in our system. Did you forget your password?"

当我尝试 except exc.SQLAlchemyError 时，我得到相同的消息(尽管我想捕获特定错误而不是一揽子捕获所有错误).我也试过 exc.IntegrityError 但没有运气(虽然它存在于 API 中).

I get the same message when I tried except exc.SQLAlchemyError (although I want to catch specific errors and not a blanket catch all). I also tried exc.IntegrityError but no luck (although it exists in the API).

我的 Python 语法是否有问题，或者我需要在 SQLAlchemy 中做一些特殊的事情来捕获它?

Is there something wrong with my Python syntax, or is there something I need to do special in SQLAlchemy to catch it?

我不知道如何解决这个问题，但我对可能导致问题的原因有一些想法.也许 try 语句没有失败而是成功了，因为 SQLAlchemy 本身引发了异常，而 Pyramid 正在生成视图，因此 except IntegrityError: 永远不会被激活.或者，更有可能的是，我完全错误地发现了这个错误.

I don't know how to solve this problem but I have a few ideas of what could be causing the problem. Maybe the try statement isn't failing but succeeding because SQLAlchemy is raising the exception itself and Pyramid is generating the view so the except IntegrityError: never gets activated. Or, more likely, I'm catching this error completely wrong.

推荐答案

在 Pyramid 中，如果您已将会话(脚手架自动为您执行的操作)配置为使用 ZopeTransactionExtension，则会话直到视图执行后才刷新/提交.如果您想自己捕获视图中的任何 SQL 错误，则需要强制 flush 将 SQL 发送到引擎.DBSession.flush() 应该在 add(...) 之后执行.

In Pyramid, if you've configured your session (which the scaffold does for you automatically) to use the ZopeTransactionExtension, then session is not flushed/committed until after the view has executed. If you want to catch any SQL errors yourself in your view, you need to force a flush to send the SQL to the engine. DBSession.flush() should do it after the add(...).

更新

我用一个保存点的例子来更新这个答案，只是因为关于如何使用事务包来做到这一点的例子很少.

I'm updating this answer with an example of a savepoint just because there are very few examples around of how to do this with the transaction package.

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)

请注意，如果不使用表级锁定，即使这很容易在事务之间重复，但它至少显示了如何使用保存点.

Note that even this is susceptible to duplicates between transactions if no table-level locking is used, but it at least shows how to use a savepoint.

这篇关于尝试使用 SQLAlchemy 捕获完整性错误的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！