I have a FastAPI app and in some of the code I have the following code
from asyncpg.exceptions import UniqueViolationError
async def cache_data(data, embedding):
async with asyncsessionmaker() as session:
db_model = CacheModel(argument_call=data, embedding=embedding, model=CURRENT_EMBEDDING_MODEL)
session.add(db_model)
try:
await session.commit()
except UniqueViolationError as e:
session.rollback()
The issue is, that an UniqueViolationError
error is not caught.
If I use from sqlalchemy.exc import IntegrityError
I can catch the error and the filter like
except IntegrityError as e:
if e.orig.sqlstate=='23505': #duplicated key error
session.rollback()
but I feel it is a bit of a hack (and rather error-prone if sqlalchely decides to change the sqlstate
).
How do I catch UniqueViolationError
directly?