feat: enhance database error handling with granular exceptions (#93)

Introduce a comprehensive exception hierarchy for database errors, enabling more precise error handling across multiple database adapters. Implement integration tests to validate the correct mapping of specific database errors to the new exception types. Address known issues in existing tests and ensure compatibility with all database adapters.

Resolves #90 (Enhancement: Make exceptions better)

Author: cofin

pyproject.toml

@@ -229,7 +229,7 @@ version = "{current_version}"
 """
 
 [tool.codespell]
-ignore-words-list = "te,ECT"
+ignore-words-list = "te,ECT,SELCT"
 skip = 'uv.lock'
 
 [tool.coverage.run]

sqlspec/adapters/adbc/driver.py

@@ -9,7 +9,6 @@
 import decimal
 from typing import TYPE_CHECKING, Any, Optional, cast
 
-from adbc_driver_manager.dbapi import DatabaseError, IntegrityError, OperationalError, ProgrammingError
 from sqlglot import exp
 
 from sqlspec.adapters.adbc.data_dictionary import AdbcDataDictionary
@@ -18,7 +17,19 @@
 from sqlspec.core.parameters import ParameterStyle, ParameterStyleConfig
 from sqlspec.core.statement import SQL, StatementConfig
 from sqlspec.driver import SyncDriverAdapterBase
-from sqlspec.exceptions import MissingDependencyError, SQLParsingError, SQLSpecError
+from sqlspec.exceptions import (
+    CheckViolationError,
+    DatabaseConnectionError,
+    DataError,
+    ForeignKeyViolationError,
+    IntegrityError,
+    MissingDependencyError,
+    NotNullViolationError,
+    SQLParsingError,
+    SQLSpecError,
+    TransactionError,
+    UniqueViolationError,
+)
 from sqlspec.typing import Empty
 from sqlspec.utils.logging import get_logger
 
@@ -342,48 +353,130 @@ def __exit__(self, *_: Any) -> None:
 
 
 class AdbcExceptionHandler:
-    """Context manager for handling database exceptions."""
+    """Context manager for handling ADBC database exceptions.
+
+    ADBC propagates underlying database errors. Exception mapping
+    depends on the specific ADBC driver being used.
+    """
 
     __slots__ = ()
 
     def __enter__(self) -> None:
         return None
 
     def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        _ = exc_tb
         if exc_type is None:
             return
+        self._map_adbc_exception(exc_val)
 
-        try:
-            if issubclass(exc_type, IntegrityError):
-                e = exc_val
-                msg = f"Integrity constraint violation: {e}"
-                raise SQLSpecError(msg) from e
-            if issubclass(exc_type, ProgrammingError):
-                e = exc_val
-                error_msg = str(e).lower()
-                if "syntax" in error_msg or "parse" in error_msg:
-                    msg = f"SQL syntax error: {e}"
-                    raise SQLParsingError(msg) from e
-                msg = f"Programming error: {e}"
-                raise SQLSpecError(msg) from e
-            if issubclass(exc_type, OperationalError):
-                e = exc_val
-                msg = f"Operational error: {e}"
-                raise SQLSpecError(msg) from e
-            if issubclass(exc_type, DatabaseError):
-                e = exc_val
-                msg = f"Database error: {e}"
-                raise SQLSpecError(msg) from e
-        except ImportError:
-            pass
-        if issubclass(exc_type, Exception):
-            e = exc_val
-            error_msg = str(e).lower()
-            if "parse" in error_msg or "syntax" in error_msg:
-                msg = f"SQL parsing failed: {e}"
-                raise SQLParsingError(msg) from e
-            msg = f"Unexpected database operation error: {e}"
-            raise SQLSpecError(msg) from e
+    def _map_adbc_exception(self, e: Any) -> None:
+        """Map ADBC exception to SQLSpec exception.
+
+        ADBC drivers may expose SQLSTATE codes or driver-specific codes.
+
+        Args:
+            e: ADBC exception instance
+        """
+        sqlstate = getattr(e, "sqlstate", None)
+
+        if sqlstate:
+            self._map_sqlstate_exception(e, sqlstate)
+        else:
+            self._map_message_based_exception(e)
+
+    def _map_sqlstate_exception(self, e: Any, sqlstate: str) -> None:
+        """Map SQLSTATE code to exception.
+
+        Args:
+            e: Exception instance
+            sqlstate: SQLSTATE error code
+        """
+        if sqlstate == "23505":
+            self._raise_unique_violation(e)
+        elif sqlstate == "23503":
+            self._raise_foreign_key_violation(e)
+        elif sqlstate == "23502":
+            self._raise_not_null_violation(e)
+        elif sqlstate == "23514":
+            self._raise_check_violation(e)
+        elif sqlstate.startswith("23"):
+            self._raise_integrity_error(e)
+        elif sqlstate.startswith("42"):
+            self._raise_parsing_error(e)
+        elif sqlstate.startswith("08"):
+            self._raise_connection_error(e)
+        elif sqlstate.startswith("40"):
+            self._raise_transaction_error(e)
+        elif sqlstate.startswith("22"):
+            self._raise_data_error(e)
+        else:
+            self._raise_generic_error(e)
+
+    def _map_message_based_exception(self, e: Any) -> None:
+        """Map exception using message-based detection.
+
+        Args:
+            e: Exception instance
+        """
+        error_msg = str(e).lower()
+
+        if "unique" in error_msg or "duplicate" in error_msg:
+            self._raise_unique_violation(e)
+        elif "foreign key" in error_msg:
+            self._raise_foreign_key_violation(e)
+        elif "not null" in error_msg or "null value" in error_msg:
+            self._raise_not_null_violation(e)
+        elif "check constraint" in error_msg:
+            self._raise_check_violation(e)
+        elif "constraint" in error_msg:
+            self._raise_integrity_error(e)
+        elif "syntax" in error_msg:
+            self._raise_parsing_error(e)
+        elif "connection" in error_msg or "connect" in error_msg:
+            self._raise_connection_error(e)
+        else:
+            self._raise_generic_error(e)
+
+    def _raise_unique_violation(self, e: Any) -> None:
+        msg = f"ADBC unique constraint violation: {e}"
+        raise UniqueViolationError(msg) from e
+
+    def _raise_foreign_key_violation(self, e: Any) -> None:
+        msg = f"ADBC foreign key constraint violation: {e}"
+        raise ForeignKeyViolationError(msg) from e
+
+    def _raise_not_null_violation(self, e: Any) -> None:
+        msg = f"ADBC not-null constraint violation: {e}"
+        raise NotNullViolationError(msg) from e
+
+    def _raise_check_violation(self, e: Any) -> None:
+        msg = f"ADBC check constraint violation: {e}"
+        raise CheckViolationError(msg) from e
+
+    def _raise_integrity_error(self, e: Any) -> None:
+        msg = f"ADBC integrity constraint violation: {e}"
+        raise IntegrityError(msg) from e
+
+    def _raise_parsing_error(self, e: Any) -> None:
+        msg = f"ADBC SQL parsing error: {e}"
+        raise SQLParsingError(msg) from e
+
+    def _raise_connection_error(self, e: Any) -> None:
+        msg = f"ADBC connection error: {e}"
+        raise DatabaseConnectionError(msg) from e
+
+    def _raise_transaction_error(self, e: Any) -> None:
+        msg = f"ADBC transaction error: {e}"
+        raise TransactionError(msg) from e
+
+    def _raise_data_error(self, e: Any) -> None:
+        msg = f"ADBC data error: {e}"
+        raise DataError(msg) from e
+
+    def _raise_generic_error(self, e: Any) -> None:
+        msg = f"ADBC database error: {e}"
+        raise SQLSpecError(msg) from e
 
 
 class AdbcDriver(SyncDriverAdapterBase):

sqlspec/adapters/aiosqlite/driver.py

@@ -12,7 +12,18 @@
 from sqlspec.core.parameters import ParameterStyle, ParameterStyleConfig
 from sqlspec.core.statement import StatementConfig
 from sqlspec.driver import AsyncDriverAdapterBase
-from sqlspec.exceptions import SQLParsingError, SQLSpecError
+from sqlspec.exceptions import (
+    CheckViolationError,
+    DatabaseConnectionError,
+    DataError,
+    ForeignKeyViolationError,
+    IntegrityError,
+    NotNullViolationError,
+    OperationalError,
+    SQLParsingError,
+    SQLSpecError,
+    UniqueViolationError,
+)
 from sqlspec.utils.serializers import to_json
 
 if TYPE_CHECKING:
@@ -26,6 +37,15 @@
 
 __all__ = ("AiosqliteCursor", "AiosqliteDriver", "AiosqliteExceptionHandler", "aiosqlite_statement_config")
 
+SQLITE_CONSTRAINT_UNIQUE_CODE = 2067
+SQLITE_CONSTRAINT_FOREIGNKEY_CODE = 787
+SQLITE_CONSTRAINT_NOTNULL_CODE = 1811
+SQLITE_CONSTRAINT_CHECK_CODE = 531
+SQLITE_CONSTRAINT_CODE = 19
+SQLITE_CANTOPEN_CODE = 14
+SQLITE_IOERR_CODE = 10
+SQLITE_MISMATCH_CODE = 20
+
 
 aiosqlite_statement_config = StatementConfig(
     dialect="sqlite",
@@ -74,7 +94,11 @@ async def __aexit__(self, *_: Any) -> None:
 
 
 class AiosqliteExceptionHandler:
-    """Async context manager for AIOSQLite database exceptions."""
+    """Async context manager for handling aiosqlite database exceptions.
+
+    Maps SQLite extended result codes to specific SQLSpec exceptions
+    for better error handling in application code.
+    """
 
     __slots__ = ()
 
@@ -84,38 +108,103 @@ async def __aenter__(self) -> None:
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         if exc_type is None:
             return
-        if issubclass(exc_type, aiosqlite.IntegrityError):
-            e = exc_val
-            msg = f"AIOSQLite integrity constraint violation: {e}"
-            raise SQLSpecError(msg) from e
-        if issubclass(exc_type, aiosqlite.OperationalError):
-            e = exc_val
-            error_msg = str(e).lower()
-            if "locked" in error_msg:
-                msg = f"AIOSQLite database locked: {e}. Consider enabling WAL mode or reducing concurrency."
-                raise SQLSpecError(msg) from e
-            if "syntax" in error_msg or "malformed" in error_msg:
-                msg = f"AIOSQLite SQL syntax error: {e}"
-                raise SQLParsingError(msg) from e
-            msg = f"AIOSQLite operational error: {e}"
-            raise SQLSpecError(msg) from e
-        if issubclass(exc_type, aiosqlite.DatabaseError):
-            e = exc_val
-            msg = f"AIOSQLite database error: {e}"
-            raise SQLSpecError(msg) from e
         if issubclass(exc_type, aiosqlite.Error):
-            e = exc_val
-            msg = f"AIOSQLite error: {e}"
-            raise SQLSpecError(msg) from e
-        if issubclass(exc_type, Exception):
-            e = exc_val
-            error_msg = str(e).lower()
-            if "parse" in error_msg or "syntax" in error_msg:
-                msg = f"SQL parsing failed: {e}"
-                raise SQLParsingError(msg) from e
-            msg = f"Unexpected async database operation error: {e}"
+            self._map_sqlite_exception(exc_val)
+
+    def _map_sqlite_exception(self, e: Any) -> None:
+        """Map SQLite exception to SQLSpec exception.
+
+        Args:
+            e: aiosqlite.Error instance
+
+        Raises:
+            Specific SQLSpec exception based on error code
+        """
+        error_code = getattr(e, "sqlite_errorcode", None)
+        error_name = getattr(e, "sqlite_errorname", None)
+        error_msg = str(e).lower()
+
+        if "locked" in error_msg:
+            msg = f"AIOSQLite database locked: {e}. Consider enabling WAL mode or reducing concurrency."
             raise SQLSpecError(msg) from e
 
+        if not error_code:
+            if "unique constraint" in error_msg:
+                self._raise_unique_violation(e, 0)
+            elif "foreign key constraint" in error_msg:
+                self._raise_foreign_key_violation(e, 0)
+            elif "not null constraint" in error_msg:
+                self._raise_not_null_violation(e, 0)
+            elif "check constraint" in error_msg:
+                self._raise_check_violation(e, 0)
+            elif "syntax" in error_msg:
+                self._raise_parsing_error(e, None)
+            else:
+                self._raise_generic_error(e)
+            return
+
+        if error_code == SQLITE_CONSTRAINT_UNIQUE_CODE or error_name == "SQLITE_CONSTRAINT_UNIQUE":
+            self._raise_unique_violation(e, error_code)
+        elif error_code == SQLITE_CONSTRAINT_FOREIGNKEY_CODE or error_name == "SQLITE_CONSTRAINT_FOREIGNKEY":
+            self._raise_foreign_key_violation(e, error_code)
+        elif error_code == SQLITE_CONSTRAINT_NOTNULL_CODE or error_name == "SQLITE_CONSTRAINT_NOTNULL":
+            self._raise_not_null_violation(e, error_code)
+        elif error_code == SQLITE_CONSTRAINT_CHECK_CODE or error_name == "SQLITE_CONSTRAINT_CHECK":
+            self._raise_check_violation(e, error_code)
+        elif error_code == SQLITE_CONSTRAINT_CODE or error_name == "SQLITE_CONSTRAINT":
+            self._raise_integrity_error(e, error_code)
+        elif error_code == SQLITE_CANTOPEN_CODE or error_name == "SQLITE_CANTOPEN":
+            self._raise_connection_error(e, error_code)
+        elif error_code == SQLITE_IOERR_CODE or error_name == "SQLITE_IOERR":
+            self._raise_operational_error(e, error_code)
+        elif error_code == SQLITE_MISMATCH_CODE or error_name == "SQLITE_MISMATCH":
+            self._raise_data_error(e, error_code)
+        elif error_code == 1 or "syntax" in error_msg:
+            self._raise_parsing_error(e, error_code)
+        else:
+            self._raise_generic_error(e)
+
+    def _raise_unique_violation(self, e: Any, code: int) -> None:
+        msg = f"SQLite unique constraint violation [code {code}]: {e}"
+        raise UniqueViolationError(msg) from e
+
+    def _raise_foreign_key_violation(self, e: Any, code: int) -> None:
+        msg = f"SQLite foreign key constraint violation [code {code}]: {e}"
+        raise ForeignKeyViolationError(msg) from e
+
+    def _raise_not_null_violation(self, e: Any, code: int) -> None:
+        msg = f"SQLite not-null constraint violation [code {code}]: {e}"
+        raise NotNullViolationError(msg) from e
+
+    def _raise_check_violation(self, e: Any, code: int) -> None:
+        msg = f"SQLite check constraint violation [code {code}]: {e}"
+        raise CheckViolationError(msg) from e
+
+    def _raise_integrity_error(self, e: Any, code: int) -> None:
+        msg = f"SQLite integrity constraint violation [code {code}]: {e}"
+        raise IntegrityError(msg) from e
+
+    def _raise_parsing_error(self, e: Any, code: "Optional[int]") -> None:
+        code_str = f"[code {code}]" if code else ""
+        msg = f"SQLite SQL syntax error {code_str}: {e}"
+        raise SQLParsingError(msg) from e
+
+    def _raise_connection_error(self, e: Any, code: int) -> None:
+        msg = f"SQLite connection error [code {code}]: {e}"
+        raise DatabaseConnectionError(msg) from e
+
+    def _raise_operational_error(self, e: Any, code: int) -> None:
+        msg = f"SQLite operational error [code {code}]: {e}"
+        raise OperationalError(msg) from e
+
+    def _raise_data_error(self, e: Any, code: int) -> None:
+        msg = f"SQLite data error [code {code}]: {e}"
+        raise DataError(msg) from e
+
+    def _raise_generic_error(self, e: Any) -> None:
+        msg = f"SQLite database error: {e}"
+        raise SQLSpecError(msg) from e
+
 
 class AiosqliteDriver(AsyncDriverAdapterBase):
     """AIOSQLite driver for async SQLite database operations."""