refactor: deprecate usage of cursor.execute statements in favor of the in class implementation of execute.

Author: gmcrocetti

pandas/io/sql.py

@@ -997,7 +997,7 @@ def _execute_insert(self, conn, keys: list[str], data_iter) -> int:
            Each item contains a list of values to be inserted
         """
         data = [dict(zip(keys, row)) for row in data_iter]
-        result = conn.execute(self.table.insert(), data)
+        result = self.pd_sql.execute(self.table.insert(), data)
         return result.rowcount
 
     def _execute_insert_multi(self, conn, keys: list[str], data_iter) -> int:
@@ -1014,7 +1014,7 @@ def _execute_insert_multi(self, conn, keys: list[str], data_iter) -> int:
 
         data = [dict(zip(keys, row)) for row in data_iter]
         stmt = insert(self.table).values(data)
-        result = conn.execute(stmt)
+        result = self.pd_sql.execute(stmt)
         return result.rowcount
 
     def insert_data(self) -> tuple[list[str], list[np.ndarray]]:
@@ -1651,10 +1651,18 @@ def run_transaction(self):
 
     def execute(self, sql: str | Select | TextClause, params=None):
         """Simple passthrough to SQLAlchemy connectable"""
+        from sqlalchemy.exc import DBAPIError as SQLAlchemyDatabaseError
+
         args = [] if params is None else [params]
         if isinstance(sql, str):
-            return self.con.exec_driver_sql(sql, *args)
-        return self.con.execute(sql, *args)
+            execute_function = self.con.exec_driver_sql
+        else:
+            execute_function = self.con.execute
+
+        try:
+            return execute_function(sql, *args)
+        except SQLAlchemyDatabaseError as exc:
+            raise DatabaseError(f"Execution failed on sql '{sql}': {exc}") from exc
 
     def read_table(
         self,
@@ -2108,17 +2116,19 @@ def run_transaction(self):
             self.con.commit()
 
     def execute(self, sql: str | Select | TextClause, params=None):
+        from adbc_driver_manager import DatabaseError as ADBCDatabaseError
+
         if not isinstance(sql, str):
             raise TypeError("Query must be a string unless using sqlalchemy.")
         args = [] if params is None else [params]
         cur = self.con.cursor()
         try:
             cur.execute(sql, *args)
             return cur
-        except Exception as exc:
+        except ADBCDatabaseError as exc:
             try:
                 self.con.rollback()
-            except Exception as inner_exc:  # pragma: no cover
+            except ADBCDatabaseError as inner_exc:  # pragma: no cover
                 ex = DatabaseError(
                     f"Execution failed on sql: {sql}\n{exc}\nunable to rollback"
                 )
@@ -2335,6 +2345,9 @@ def to_sql(
         engine : {'auto', 'sqlalchemy'}, default 'auto'
             Raises NotImplementedError if not set to 'auto'
         """
+        from adbc_driver_manager import DatabaseError as ADBCDatabaseError
+        import pyarrow as pa
+
         if index_label:
             raise NotImplementedError(
                 "'index_label' is not implemented for ADBC drivers"
@@ -2369,17 +2382,18 @@ def to_sql(
             elif if_exists == "append":
                 mode = "append"
 
-        import pyarrow as pa
-
         try:
             tbl = pa.Table.from_pandas(frame, preserve_index=index)
         except pa.ArrowNotImplementedError as exc:
             raise ValueError("datatypes not supported") from exc
 
         with self.con.cursor() as cur:
-            total_inserted = cur.adbc_ingest(
-                table_name=name, data=tbl, mode=mode, db_schema_name=schema
-            )
+            try:
+                total_inserted = cur.adbc_ingest(
+                    table_name=name, data=tbl, mode=mode, db_schema_name=schema
+                )
+            except ADBCDatabaseError as exc:
+                raise DatabaseError("Execution failed") from exc
 
         self.con.commit()
         return total_inserted
@@ -2520,8 +2534,13 @@ def insert_statement(self, *, num_rows: int) -> str:
         return insert_statement
 
     def _execute_insert(self, conn, keys, data_iter) -> int:
+        from sqlite3 import DatabaseError as SQLiteDatabaseError
+
         data_list = list(data_iter)
-        conn.executemany(self.insert_statement(num_rows=1), data_list)
+        try:
+            conn.executemany(self.insert_statement(num_rows=1), data_list)
+        except SQLiteDatabaseError as exc:
+            raise DatabaseError("Execution failed") from exc
         return conn.rowcount
 
     def _execute_insert_multi(self, conn, keys, data_iter) -> int:

pandas/tests/io/test_sql.py

@@ -3409,8 +3409,8 @@ def test_to_sql_with_negative_npinf(conn, request, input):
             mark = pytest.mark.xfail(reason="GH 36465")
             request.applymarker(mark)
 
-        msg = "inf cannot be used with MySQL"
-        with pytest.raises(ValueError, match=msg):
+        msg = "Execution failed on sql"
+        with pytest.raises(pd.errors.DatabaseError, match=msg):
             df.to_sql(name="foobar", con=conn, index=False)
     else:
         assert df.to_sql(name="foobar", con=conn, index=False) == 1