How do I catch a psycopg2.errors.UniqueViolation error in a Python (Flask) app?

Question

I have a small Python web app (written in Flask) that uses sqlalchemy to persist data to the database. When I try to insert a duplicate row, an exception is raised, something like this:

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

I would like to wrap the exception and re-raise my own so I can add my own logging and messaging that is specific to that particular error. This is what I tried (simplified):

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

But this fails to trap the error because psycopg2.errors.UniqueViolation isn't actually a class name (!).

In PHP, this would be as easy as catching copy/pasting the classname of the exception, but in Python, this is much more obfuscated.

There was a similar question here, but it didn't deal with this specific use-case and (importantly), it did not clarify how one can identify the root exception class name.

How does one find out what exception is actually being raised? Why does Python hide this?

Answer 1

The error that you have posted in your question isn't the error that has been raised. The full error message is:

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

The key part being the SQLAlchemy error which you've chosen to omit for some reason. SQLAlchemy catches the original error, wraps it in it's own error and raises that.

but in Python, this is much more obfuscated... Why does Python hide this?

This isn't obfuscation, nothing is hidden, the behavior is documented, specific to the frameworks that you are using and is not enforced by the Python language. SQLAlchemy is an abstraction library and if it were to raise exceptions specific to the underlying dpapi adapter, it would significantly reduce the portability of code written within it.

From the docs:

SQLAlchemy does not generate these exceptions directly. Instead, they are intercepted from the database driver and wrapped by the SQLAlchemy-provided exception DBAPIError, however the messaging within the exception is generated by the driver, not SQLAlchemy.

Exceptions raised by the dbapi layer are wrapped in a subclass of the sqlalchemy.exc.DBAPIError, where it is noted:

The wrapped exception object is available in the orig attribute.

So it's very straightforward to catch the SQLAlchemy exception and inspect the original exception, which is an instance of psycopg2.errors.UniqueViolation, as you'd expect. However, unless your error handling is very specific to the type raised by the dbapi layer, I'd suggest that inspecting the underlying type might be unnecessary as the SQLAlchemy exception that is raised will provide enough runtime information to do what you have to do.

Here is an example script that raises a sqlalchemy.exc.IntegrityError, catches it, inspects the underlying exception through the orig attribute and raises an alternate, locally-defined exception.

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

And that raises:

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

According to psycopg2 docs:

In compliance with the DB API 2.0, the module makes informations about errors available through the following exceptions:

exception psycopg2.Error

Exception that is the base class of all other error exceptions. You can use this to catch all errors with one single except statement. Warnings are not considered errors and thus not use this class as base. It is a subclass of the Python StandardError (Exception on Python 3).

Thus, the proper way to catch the exceptions is:

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

Yours in particular is error 23505 according to the ErrCodes Table

Answer 3

I have a slightly different answer that avoids looking up the specific numerical error code. Simply import the constant that defines UNIQUE_VIOLATION:

from psycopg2.errorcodes import UNIQUE_VIOLATION
from psycopg2 import errors

Then use the error lookup function:

except errors.lookup(UNIQUE_VIOLATION) as e:

Solved the issue for me. You can import other error code constants as necessary.

Answer 4

For a quick reference to how to import the psycopg2 UniqueViolation (or any other error) with some quick recipes.

1) Import 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) Import IntegrityError

UniqueViolation base-exception is actually IntegrityError , so for a broader error catch (for whatever reason, normally is not recommended, but rule are meant to be broken)

   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) Source Code

The psycopg2 errors module is found here --> /psycopg2/errors.py, and is actually like a gateway for the real error code list.

Here you can see the function used to call the correct error by given code:

    #
    # 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]

But the really juicy stuff are found here ---> \psycopg2\_psycopg\__init__.py Once here find the variable sqlstate_errors which is a dict containing the codes as value and the actual error as Note, here is a small snippet (is pretty big):

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

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