From 826188455fc184bfca34ba044518a74e4989c92c Mon Sep 17 00:00:00 2001 From: Konstantin Fadeev Date: Tue, 28 Apr 2026 20:15:45 +0300 Subject: [PATCH] Add handlers for specific errors. --- CHANGES.md | 4 +++ pyproject.toml | 2 +- src/db_first/dbal/sqla.py | 51 +++++++++++++++++++++-------------- tests/dbal/test_dbal_error.py | 8 +++--- 4 files changed, 40 insertions(+), 25 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index d5400e8..70c1aa9 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,3 +1,7 @@ +## Version 5.3.1 + +* Add handlers for specific errors. + ## Version 5.3.0 * Create specific exceptions for errors. diff --git a/pyproject.toml b/pyproject.toml index 2b41012..bd323e5 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/src/db_first/dbal/sqla.py b/src/db_first/dbal/sqla.py index 9d1b107..80f7843 100644 --- a/src/db_first/dbal/sqla.py +++ b/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) diff --git a/tests/dbal/test_dbal_error.py b/tests/dbal/test_dbal_error.py index 16e013a..ff3173f 100644 --- a/tests/dbal/test_dbal_error.py +++ b/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]}])