19

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

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 中）。

我的 Python 语法有什么问题，还是我需要在 SQLAlchemy 中做一些特别的事情来捕捉它？

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

4

6 回答 6

23

在 Pyramid 中，如果您已将会话（脚手架自动为您执行）配置为使用ZopeTransactionExtension，则在视图执行之前不会刷新/提交会话。如果您想自己在视图中捕获任何 SQL 错误，则需要强制 aflush将 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)

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

于 2012-07-03T17:27:59.797 回答
8

您需要做的是捕获一般异常并输出其类；那么您可以使异常更具体。

except Exception as ex:
    print ex.__class__
于 2012-07-03T16:23:02.070 回答
5

可能没有数据库操作，直到DBSession.commit()在堆栈中稍后在已经返回IntegrityError的控制器代码之后引发。try/except

于 2012-07-03T16:49:49.857 回答
1

编辑：上面编辑的答案是使用回滚的更好方法。

--

如果您想在金字塔应用程序的中间处理事务或在序列末尾执行自动事务提交的事情，那么不需要发生任何魔法。

如果前一个事务失败，请记住启动一个新事务。

像这样：

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，您可以使用保存点并回滚到保存点而不是开始新事务；遗憾的是，这是不可能的，因为尝试提交会使已知的先前保存点无效。好东西啊？） （编辑：<---原来我错了......）

于 2013-09-16T15:46:57.170 回答
1

我就是这样做的。

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-关闭它

于 2015-05-20T21:22:56.597 回答
0

刷新会话后在 finally 语句中捕获异常。

try:
    new_user = Users(email, firstname, lastname, password)
    DBSession.add(new_user)
    return HTTPFound(location = request.route_url('new'))
finally:
    try:
        DBSession.flush()
    except IntegrityError:
        message1 = "Yikes! Your email already exists in our system. Did you forget your password?"
于 2022-02-26T15:50:17.527 回答