How to find the offending attribute with a sqlalchemy IntegrityError

Question

I have a very simple SqlAlchemy model

class User(Base):
    """ The SQLAlchemy declarative model class for a User object. """
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    phone = Column(String, unique=True)
    email = Column(String, unique=True)

When inserting a new User, an IntegrityError could occur if the email or phone is a duplicate.

Is there any way to detect which of the columns was violating the integrity error? Or is the only way to do a separate query to see or a value is present?

Motley

You could enclose each in a try/except clause catching the IntegrityError and reacting accordingly...But isn't it possible for 2 people to share a phone? (my wife and I do...) – Dreamy

@Dreamy Yes, I catch it, but I want to know which of the two columns is the violating one; email or phone. And for this business case it is needed that phone numbers are unique (logging in by phone). – Motley

I do not think you can do that without parsing a message of the exception provided by the backend (or rdbms driver) used. – Mensa

Answer 1

You can use the below way to get the underlying code, message, and format the message accordingly.

except exc.IntegrityError as e:
       errorInfo = e.orig.args
       print(errorInfo[0])  #This will give you error code
       print(errorInfo[1])  #This will give you error message

BTW, you have to import exc from sqlalchemy: from sqlalchemy import exc Let me know if you need any other info. I can try it out.

For more info on sqlalchemy exc, please find the code: https://github.com/zzzeek/sqlalchemy/blob/master/lib/sqlalchemy/exc.py

Regorge

The info you get from e.orig.args is dependent on what database engine you're running. For example, sqlite results in a tuple of size 1 and doesn't have an error code. – Charin

Answer 2

try:
    ....
except IntegrityError as e:
    print(e.orig.diag.message_detail)

This worked for me.

Grin

Exactly what I was looking for. Thanks! – Himalayas

Answer 3

There's no clean way to do this unfortunately but I use the orig attribute on the IntegrityError and the parse module:

try:
    db.session.add(user)
    db.session.commit()
except IntegrityError, e:
    dupe_field = parse('duplicate key value violates unique constraint "{constraint}"\nDETAIL:  Key ({field})=({input}) already exists.\n', str(e.orig))["field"]

This may not be the only error string IntegrityError throws and it could change in future updates to SQLAlchemy so its not ideal

Nomology

what parse module is this? – Selfabuse

@AsavPatel Here you go pypi.org/project/parse – Indies

Answer 4

When a database (PostgreSQL, SQLite, etc.) throws an error, SQLAlchemy wraps it in a new exception of type IntegrityError.

You can always get the original exception that was raised by the database using e.orig:

from sqlalchemy.exc import IntegrityError

try:
    ...
except IntegrityError as e:
    print(e.orig)

You can now use e.orig to check the exact constraint that was violated. For example, in PostgreSQL, you can use e.orig.diag.constraint_name.

Let's assume the phone unique constraint of the users table was violated, then the constraint name will be called something like: users_phone_key.

try:
    ...
except IntegrityError as e:
    if e.orig.diag.constraint_name == "users_phone_key":
        print("Phone number is already used")
    elif e.orig.diag.constraint_name == "users_email_key":
        print("Email is already used")

PS: as @cacheoff mentioned, you can use message_detail instead of constraint_name to get a more readable error message.

Almswoman

Answer 5

Same solution as @pixelperfect but using standard library (re module)

def get_conflicting_field(err: sqlalchemy.exc.IntegrityError) -> tuple[str, str] | None:
    """
    Parses the IntegrityError message and returns tuple with conflicting field name and value.
    """
    pattern = re.compile(r'DETAIL\:\s+Key \((?P<field>.+?)\)=\((?P<value>.+?)\) already exists')
    match = pattern.search(str(err))
    if match is not None:
        return match['field'], match['value']

Niall

Answer 6

I normally use a try catch for this.

try:
    session.commit()
catch:   
str(sys.exc_info()[0]) + " \nDESCRIPTION:  "+ str(sys.exc_info()[1]) + "\n" + str(sys.exc_info()[2])

When I encounter an integrity error, I get the following message and I skip that particular transcation and continue with the rest of them

DESCRIPTION:  (IntegrityError) duplicate key value violates unique constraint "test_code"
DETAIL:  Key (test_code)=(5342) already exists.
'INSERT INTO test_table (pk, test_code, test_name) VALUES (%(pk)s, %(test_code)s, %(test_name)s)' { 'pk': '1', 'test_code': '5342', 'test_name': 'test' }

Unworldly