如何在 Python (Flask) 应用程序中捕获 psycopg2.errors.UniqueViolation 错误？

Question

我有一个小型 Python Web 应用程序（用 Flask 编写），它使用 sqlalchemy 将数据持久保存到数据库中。当我尝试插入重复行时，会引发异常，如下所示：

(psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "uix_my_column"

我想包装异常并重新引发我自己的，这样我就可以添加自己的日志记录和特定于该特定错误的消息传递。这是我尝试过的（简化）：

from db import DbApi
from my_exceptions import BadRequest
from psycopg2.errors import UniqueViolation # <-- this does not exist!

class MyClass:
    def __init__(self):
        self.db = DbApi() 

    def create(self, data: dict) -> MyRecord:
        try:
            with self.db.session_local(expire_on_commit=False) as session:
                my_rec = MyRecord(**data)
                session.add(my_rec)
                session.commit()
                session.refresh(my_rec)
                return my_rec
        except UniqueViolation as e:
            raise BadRequest('A duplicate record already exists')

但这无法捕获错误，因为 psycopg2.errors.UniqueViolation 实际上不是类名 (!)。

在 PHP 中，这就像捕获复制/粘贴异常的类名一样简单，但在 Python 中，这更加模糊。

有一个类似的问题here，但它没有处理这个特定的用例并且（重要的是）它没有阐明如何识别根异常类名称。

如何找出实际引发的异常？ Python 为什么要隐藏这个？

Answer 1

您在问题中发布的错误不是已引发的错误。完整的错误信息是：

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "model_name_key"

关键部分是您出于某种原因选择省略的 SQLAlchemy 错误。 SQLAlchemy 捕获原始错误，将其包装在它自己的错误中并引发它。

但在 Python 中，这更加模糊......为什么 Python 隐藏这个？

这不是混淆，没有任何隐藏，行为被记录，特定于您正在使用的框架，并且不是由 Python 语言强制执行的。 SQLAlchemy 是一个抽象库，如果它引发特定于底层 dpapi 适配器的异常，则会显着降低其中编写的代码的可移植性。

来自docs：

SQLAlchemy 不会直接生成这些异常。相反，他们 从数据库驱动程序中截取并由 SQLAlchemy 提供的异常 DBAPIError，但是其中的消息传递 异常是由驱动程序生成的，而不是 SQLAlchemy。

dbapi 层引发的异常被包装在sqlalchemy.exc.DBAPIError 的子类中，在此注明：

封装的异常对象在orig 属性中可用。

因此，正如您所料，捕获 SQLAlchemy 异常并检查原始异常（psycopg2.errors.UniqueViolation 的一个实例）非常简单。但是，除非您的错误处理非常特定于 dbapi 层引发的类型，否则我建议检查底层类型可能是不必要的，因为引发的 SQLAlchemy 异常将提供足够的运行时信息来完成您必须做的事情。

这是一个示例脚本，它引发 sqlalchemy.exc.IntegrityError，捕获它，通过 orig 属性检查底层异常并引发一个替代的本地定义异常。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from psycopg2.errors import UniqueViolation


engine = create_engine("postgresql+psycopg2://some-user:mysecretpassword@localhost:5432/some-user")

Base = declarative_base()
Session = sessionmaker(bind=engine)


class BadRequest(Exception):
    pass


class Model(Base):
    __tablename__ = "model"
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)


if __name__ == "__main__":
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    s = Session()
    s.add(Model(name="a"))
    s.commit()
    s.add(Model(name="a"))
    try:
        s.commit()
    except IntegrityError as e:
        assert isinstance(e.orig, UniqueViolation)  # proves the original exception
        raise BadRequest from e

这引发了：

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "model_name_key"
DETAIL:  Key (name)=(a) already exists.

[SQL: INSERT INTO model (name) VALUES (%(name)s) RETURNING model.id]
[parameters: {'name': 'a'}]
(Background on this error at: http://sqlalche.me/e/gkpj)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File ".\main.py", line 36, in <module>
    raise BadRequest from e
__main__.BadRequest

Answer 2

根据 psycopg2 docs:

根据 DB API 2.0，该模块通过以下异常提供有关错误的信息：

异常 psycopg2.Error

异常是所有其他错误异常的基类。您可以使用它通过一个 except 语句来捕获所有错误。警告不被视为错误，因此不使用此类作为基础。它是 Python StandardError 的子类（Python 3 上的异常）。

因此，捕获异常的正确方法是：

try:
    # your stuff here
except psycopg2.Error as e:
    # get error code
    error = e.pgcode
    # then do something.

根据ErrCodes Table，尤其是您的错误 23505

Answer 3

我有一个稍微不同的答案，可以避免查找特定的数字错误代码。只需导入定义UNIQUE_VIOLATION 的常量：

from psycopg2.errorcodes import UNIQUE_VIOLATION
from psycopg2 import errors

然后使用错误查找功能：

except errors.lookup(UNIQUE_VIOLATION) as e:

为我解决了这个问题。您可以根据需要导入其他错误代码常量。

Answer 4

快速参考如何导入 psycopg2 UniqueViolation（或任何其他错误） 有一些快速食谱。

1) 导入 UniqueViolation

   import traceback   # Used for printing the full traceback | Better for debug.
   from psycopg2 import errors

   UniqueViolation = errors.lookup('23505')  # Correct way to Import the psycopg2 errors

   # ...... Code ....

    try:
        db.commit()           
    except UniqueViolation as err:
        traceback.print_exc()
        db.rollback()            

    # ...... Code ....

---

2) 导入完整性错误

UniqueViolation 基本异常实际上是 IntegrityError ，因此对于更广泛的错误捕获 （无论出于何种原因，通常不推荐，但规则是要被打破的）

   import traceback   # Used for printing the full traceback | Better for debug.
   from psycopg2._psycopg import IntegrityError

   # ...... Code ....

    try:
        db.commit()           
    except IntegrityError as err:
        traceback.print_exc()
        db.rollback()            

    # ...... Code ....

---

3) 源代码

psycopg2 错误模块可在此处找到 --> /psycopg2/errors.py，它实际上就像是真正的错误代码列表的网关。

在这里您可以看到用于通过给定代码调用正确错误的函数：

    #
    # NOTE: the exceptions are injected into this module by the C extention.
    #
    
    
    def lookup(code):
        """Lookup an error code and return its exception class.
    
        Raise `!KeyError` if the code is not found.
        """
        from psycopg2._psycopg import sqlstate_errors   # avoid circular import
        return sqlstate_errors[code]

但真正多汁的东西在这里 ---> \psycopg2\_psycopg\__init__.py 一旦在这里找到变量sqlstate_errors，它是一个包含代码作为值和实际错误作为注释的字典，这里是一个小的sn-p（非常大）：

    sqlstate_errors = {
        '02000': None, # (!) real value is "<class 'psycopg2.errors.NoData'>"
        '02001': None, # (!) real value is "<class 'psycopg2.errors.NoAdditionalDynamicResultSetsReturned'>"
        '03000': None, # (!) real value is "<class 'psycopg2.errors.SqlStatementNotYetComplete'>"
        '08000': None, # (!) real value is "<class 'psycopg2.errors.ConnectionException'>"
        '08001': None, # (!) real value is "<class 'psycopg2.errors.SqlclientUnableToEstablishSqlconnection'>"
        '08003': None, # (!) real value is "<class 'psycopg2.errors.ConnectionDoesNotExist'>"
        '08004': None, # (!) real value is "<class 'psycopg2.errors.SqlserverRejectedEstablishmentOfSqlconnection'>"
        '08006': None, # (!) real value is "<class 'psycopg2.errors.ConnectionFailure'>"
        '08007': None, # (!) real value is "<class 'psycopg2.errors.TransactionResolutionUnknown'>"
        '08P01': None, # (!) real value is "<class 'psycopg2.errors.ProtocolViolation'>"
 # -------- Lots of lines ---------- # 
        '23503': None, # (!) real value is "<class 'psycopg2.errors.ForeignKeyViolation'>"
        # There you are!!!
        '23505': None, # (!) real value is "<class 'psycopg2.errors.UniqueViolation'>" 
        # ---------------- 
        '23514': None, # (!) real value is "<class 'psycopg2.errors.CheckViolation'>"
        '23P01': None, # (!) real value is "<class 'psycopg2.errors.ExclusionViolation'>"

---

4) 文档

• psycopg.org | SQLSTATE Exception
• postgresq | PostgreSQL Error Codes

Answer 5

from db import DbApi
from my_exceptions import BadRequest
from psycopg2 import errors

class MyClass:
    def __init__(self):
        self.db = DbApi() 

    def create(self, data: dict) -> MyRecord:
        try:
            with self.db.session_local(expire_on_commit=False) as session:
                my_rec = MyRecord(**data)
                session.add(my_rec)
                session.commit()
                session.refresh(my_rec)
                return my_rec
        except errors.lookup("23505"):
            raise BadRequest('A duplicate record already exists')

我收到 psycopg2.errors.UniqueViolation 错误。我该如何处理？

有关如何处理 psycopg2 错误的更多详细信息，请参阅https://www.psycopg.org/docs/errors.html