Add handlers for specific errors.

CHANGES.md

@@ -1,3 +1,7 @@
+## Version 5.3.1
+
+* Add handlers for specific errors.
+
 ## Version 5.3.0
 
 * Create specific exceptions for errors.

pyproject.toml

@@ -25,7 +25,7 @@ license = {file = "LICENSE"}
 name = "DB-First"
 readme = "README.md"
 requires-python = ">=3.11"
-version = "5.3.0"
+version = "5.3.1"
 
 [project.optional-dependencies]
 dev = [

src/db_first/dbal/sqla.py

@@ -24,6 +24,28 @@
 from sqlalchemy.orm.exc import StaleDataError
 
 
+def compile_error_handler(e: CompileError) -> None:
+    if e.args[0].startswith('Unconsumed column names:'):
+        raise DBALColumnNonExistException(e)
+
+
+def integrity_error_handler(e: IntegrityError, s: Session) -> None:
+    db_type = s.get_bind().name
+
+    if db_type == 'sqlite':
+        if e.orig.sqlite_errorname == 'SQLITE_CONSTRAINT_NOTNULL':
+            raise DBALNotNullConstraintFailedException(e.orig.args[0])
+        if e.orig.sqlite_errorname == 'SQLITE_CONSTRAINT_FOREIGNKEY':
+            raise DBALForeignKeyConstraintFailedException(e.orig.args[0])
+
+    elif db_type == 'postgresql':
+        if 'is not present in table' in e.orig.diag.message_detail:
+            raise DBALForeignKeyConstraintFailedException(e.orig.diag.message_detail)
+
+    else:
+        raise NotImplementedError(f'DB <{db_type}> not implemented.')
+
+
 class SqlaDBAL[M](PageMixin):
     """Base SqlaDBAL, implement base CRUD sqlalchemy operations."""
 
@@ -48,19 +70,13 @@ def create(self, **kwargs) -> M:
 
         except CompileError as e:
             self._session.rollback()
-            if e.args[0].startswith('Unconsumed column names:'):
-                raise DBALColumnNonExistException(e)
-            else:
-                raise DBALCreateException(e)
+            compile_error_handler(e)
+            raise DBALCreateException(e)
 
         except IntegrityError as e:
             self._session.rollback()
-            if e.orig.args[0].startswith('NOT NULL constraint failed: '):
-                raise DBALNotNullConstraintFailedException(e)
-            elif e.orig.args[0].startswith('FOREIGN KEY constraint failed'):
-                raise DBALForeignKeyConstraintFailedException(e)
-            else:
-                raise DBALCreateException(e)
+            integrity_error_handler(e, self._session)
+            raise DBALCreateException(e)
 
         except ProgrammingError as e:
             self._session.rollback()
@@ -130,18 +146,13 @@ def update(self, id: Any, **data) -> M:
         try:
             obj = self._session.scalars(stmt).one()
         except CompileError as e:
-            if e.args[0].startswith('Unconsumed column names:'):
-                raise DBALColumnNonExistException(e)
-            else:
-                raise DBALUpdateException(e)
+            compile_error_handler(e)
+            raise DBALUpdateException(e)
 
         except IntegrityError as e:
-            if e.orig.args[0].startswith('NOT NULL constraint failed: '):
-                raise DBALNotNullConstraintFailedException(e)
-            elif e.orig.args[0].startswith('FOREIGN KEY constraint failed'):
-                raise DBALForeignKeyConstraintFailedException(e)
-            else:
-                raise DBALUpdateException(e)
+            self._session.rollback()
+            integrity_error_handler(e, self._session)
+            raise DBALUpdateException(e)
 
         except ProgrammingError as e:
             raise DBALUnexpectedValueTypeException(e)

tests/dbal/test_dbal_error.py

@@ -49,7 +49,7 @@ def test_dbal__create_error__not_null_constraint_failed(fx_db, fx_parent_dbal):
     with pytest.raises(DBALNotNullConstraintFailedException) as e:
         fx_parent_dbal(session_db).create(**data)
 
-    assert e.value.args[0].orig.args[0] == 'NOT NULL constraint failed: parents.first'
+    assert e.value.args[0] == 'NOT NULL constraint failed: parents.first'
 
 
 def test_dbal__create_error__foreign_key_constraint_failed(fx_db, fx_parent_dbal):
@@ -60,7 +60,7 @@ def test_dbal__create_error__foreign_key_constraint_failed(fx_db, fx_parent_dbal
     with pytest.raises(DBALForeignKeyConstraintFailedException) as e:
         fx_parent_dbal(session_db).create(**data)
 
-    assert e.value.args[0].orig.args[0] == 'FOREIGN KEY constraint failed'
+    assert e.value.args[0] == 'FOREIGN KEY constraint failed'
 
 
 @pytest.mark.parametrize('data', [{'first': [1, 2, 3]}])
@@ -123,7 +123,7 @@ def test_dbal__update_error__not_null_constraint_failed(fx_db, fx_parent_dbal):
     with pytest.raises(DBALNotNullConstraintFailedException) as e:
         fx_parent_dbal(session_db).update(new.id, **updated_params)
 
-    assert e.value.args[0].orig.args[0] == 'NOT NULL constraint failed: parents.first'
+    assert e.value.args[0] == 'NOT NULL constraint failed: parents.first'
 
 
 def test_dbal__update_error__foreign_key_constraint_failed(fx_db, fx_parent_dbal):
@@ -137,7 +137,7 @@ def test_dbal__update_error__foreign_key_constraint_failed(fx_db, fx_parent_dbal
     with pytest.raises(DBALForeignKeyConstraintFailedException) as e:
         fx_parent_dbal(session_db).update(new.id, **updated_params)
 
-    assert e.value.args[0].orig.args[0] == 'FOREIGN KEY constraint failed'
+    assert e.value.args[0] == 'FOREIGN KEY constraint failed'
 
 
 @pytest.mark.parametrize('data', [{'first': [1, 2, 3]}])