How to differentiate causes of SQLAlchemy's IntegrityError?
SQLAlchemy appears to just throw a general IntegrityError when there is a data integrity problem with a transaction. Of course, the exact query and error message are contained in the exception, which is sufficient for a human debugging the program. However, when writing error handling code for the exception, there doesn't seem to be a good way so far as I can tell to check which constraint on which table was responsible for the error. Also, the exception is raised by the session.commit() line rather than the line actually responsible for producing the error, so I can't differentiate using multiple try/except blocks either.
Is there a way, short of trying to programmatically parse the error message and/or query, that I can for example distinguish a duplicate primary key error from a foreign key error or a failed CHECK constraint and so forth? Or even just a way to tell which column of which table is in violation of the data integrity? Or just a way to raise the exception immediately on the line that caused the error rather than waiting for the transaction to be committed?
I ended up using session.flush() to trigger the exceptions earlier. I call it once before the line(s) in question (so I know for 100% certain that the exception wasn't triggered by previous lines) and again inside a try/catch block to see if the line(s) in question caused an error.
I admin I'm not completely happy with this solution, but I haven't been able to find anything else. I'd still love to hear if there is a better solution, ideally one that will tell me exactly which constraint of which table caused the error. But, this is a workaround that might help someone.
The IntegrityError instance has orig and statement attributes which can be inspected to obtain the error message and the failing SQL statement, respectively.
Given this model:
class Foo(Base):
__tablename__ = 'foo20201209'
id = sa.Column(sa.Integer, primary_key=True)
bar = sa.Column(sa.String(2), unique=True)
baz = sa.Column(sa.Integer, sa.CheckConstraint('baz >= 0'), default=0)
this code:
conn_strings = ['postgresql+psycopg2:///test',
'mysql+mysqlconnector:///test',
'sqlite://']
for cs in conn_strings:
engine = sa.create_engine(cs)
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
session = orm.Session(bind=engine)
for kwds in [{'bar': 'a'}, {'bar': 'a'}, {'bar': 'b', 'baz': -11}]:
session.add(Foo(**kwds))
try:
session.commit()
except sa.exc.IntegrityError as ex:
print(ex.orig)
print(ex.statement)
print()
session.rollback()
session.close()
engine.dispose()
will produce this output:
duplicate key value violates unique constraint "foo20201209_bar_key"
DETAIL: Key (bar)=(a) already exists.
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s) RETURNING foo20201209.id
new row for relation "foo20201209" violates check constraint "foo20201209_baz_check"
DETAIL: Failing row contains (3, b, -11).
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s) RETURNING foo20201209.id
1062 (23000): Duplicate entry 'a' for key 'bar'
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s)
4025 (23000): CONSTRAINT `foo20201209.baz` failed for `test`.`foo20201209`
INSERT INTO foo20201209 (bar, baz) VALUES (%(bar)s, %(baz)s)
UNIQUE constraint failed: foo20201209.bar
INSERT INTO foo20201209 (bar, baz) VALUES (?, ?)
CHECK constraint failed: foo20201209
INSERT INTO foo20201209 (bar, baz) VALUES (?, ?)
I have faced the same trouble, and it looks like, for instance, for constraints it's a better solution to give them names, parsing those exception messages then. Anyway, for my SQLAlchemy==1.4.28, a UNIQUE constraint name is still not presented anywhere in the Exception thrown.
Consider the next example:
class M(Model):
a = Column(String)
i = Column(Integer)
UniqueConstraint(M.i, name="my unique")
CheckConstraint(0 <= M.i, name="my check")
def test_test():
db = create_test_database(Model)
try:
with db.session() as s:
# Here we break our UNIQUE constraint
try:
with s.begin_nested():
s.add_all(
[
M(a="Alice", i=1),
M(a="Bob", i=1),
]
)
except IntegrityError as err:
rich.inspect(err)
# Here we break our CHECK constraint
s.add(M(a="Alice", i=-1))
except IntegrityError as err:
rich.inspect(err)
And here are the results:
┌────────────────────── <class 'sqlalchemy.exc.IntegrityError'> ───────────────────────┐
│ Wraps a DB-API IntegrityError. │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────────┐ │
│ │ IntegrityError('(sqlite3.IntegrityError) UNIQUE constraint failed: M.i') │ │
│ └──────────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ args = ('(sqlite3.IntegrityError) UNIQUE constraint failed: M.i',) │
│ code = 'gkpj' │
│ connection_invalidated = False │
│ detail = [] │
│ hide_parameters = False │
│ ismulti = False │
│ orig = IntegrityError('UNIQUE constraint failed: M.i') │
│ params = ('Bob', 1) │
│ statement = 'INSERT INTO "M" (a, i) VALUES (?, ?)' │
└──────────────────────────────────────────────────────────────────────────────────────┘
┌──────────────────────── <class 'sqlalchemy.exc.IntegrityError'> ─────────────────────────┐
│ Wraps a DB-API IntegrityError. │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────────────┐ │
│ │ IntegrityError('(sqlite3.IntegrityError) CHECK constraint failed: my check') │ │
│ └──────────────────────────────────────────────────────────────────────────────────────┘ │
│ │
│ args = ('(sqlite3.IntegrityError) CHECK constraint failed: my check',) │
│ code = 'gkpj' │
│ connection_invalidated = False │
│ detail = [] │
│ hide_parameters = False │
│ ismulti = False │
│ orig = IntegrityError('CHECK constraint failed: my check') │
│ params = ('Alice', -1) │
│ statement = 'INSERT INTO "M" (a, i) VALUES (?, ?)' │
└──────────────────────────────────────────────────────────────────────────────────────────┘
So, it looks like this CHECK constraint name will always be somewhere in the exception's string, and you can write a code around this. And it's clear why this is extra info for a UNIQUE constraint, cause you have already had the word UNIQUE and a field's name (A.i in my example). I don't think they will ever change these strings formats, but it'd be interesting to ask them about.
WARNING:
The problem is that my code relates to SQLite, while yours may relate to another DB and those messages will differ cause they are originally from the underlying DB engine, not from SQLAlchemy itself. So you have to take care about abstracting your code for those strings.