wip - trying out new solution

Author: gmcrocetti

pandas/io/sql.py

@@ -1006,7 +1006,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:
@@ -1023,7 +1023,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]]:
@@ -1670,7 +1670,7 @@ def execute(self, sql: str | Select | TextClause, params=None):
 
         try:
             return execute_function(sql, *args)
-        except SQLAlchemyError as exc:
+        except Exception as exc:
             raise DatabaseError(f"Execution failed on sql '{sql}': {exc}") from exc
 
     def read_table(
@@ -2085,9 +2085,9 @@ def delete_rows(self, table_name: str, schema: str | None = None) -> None:
             self.meta.reflect(
                 bind=self.con, only=[table_name], schema=schema, views=True
             )
-            with self.run_transaction() as con:
+            with self.run_transaction():
                 table = self.get_table(table_name, schema)
-                con.execute(table.delete())
+                self.execute(table.delete())
 
             self.meta.clear()
 
@@ -2442,8 +2442,7 @@ def has_table(self, name: str, schema: str | None = None) -> bool:
     def delete_rows(self, name: str, schema: str | None = None) -> None:
         table_name = f"{schema}.{name}" if schema else name
         if self.has_table(name, schema):
-            with self.con.cursor() as cur:
-                cur.execute(f"DELETE FROM {table_name}")
+            self.execute(f"DELETE FROM {table_name}").close()
 
     def _create_sql_schema(
         self,

pandas/tests/io/test_sql.py

@@ -2718,7 +2718,6 @@ def test_delete_rows_success(conn_name, test_frame1, request):
 
 @pytest.mark.parametrize("conn_name", all_connectable)
 def test_delete_rows_is_atomic(conn_name, request):
-    adbc_driver_manager = pytest.importorskip("adbc_driver_manager")
     sqlalchemy = pytest.importorskip("sqlalchemy")
 
     table_name = "temp_frame"
@@ -2737,20 +2736,11 @@ def test_delete_rows_is_atomic(conn_name, request):
     with pandasSQL.run_transaction() as cur:
         cur.execute(table_stmt)
 
-    if conn_name != "sqlite_buildin" and "adbc" not in conn_name:
-        expected_exception = sqlalchemy.exc.IntegrityError
-    elif "adbc" in conn_name and "sqlite" in conn_name:
-        expected_exception = adbc_driver_manager.InternalError
-    elif "adbc" in conn_name and "postgres" in conn_name:
-        expected_exception = adbc_driver_manager.ProgrammingError
-    elif conn_name == "sqlite_buildin":
-        expected_exception = sqlite3.IntegrityError
-
     with pandasSQL.run_transaction():
         pandasSQL.to_sql(original_df, table_name, if_exists="append", index=False)
 
     # inserting duplicated values in a UNIQUE constraint column
-    with pytest.raises(expected_exception):
+    with pytest.raises(pd.errors.DatabaseError):
         with pandasSQL.run_transaction():
             pandasSQL.to_sql(
                 replacing_df, table_name, if_exists="delete_rows", index=False
@@ -3473,8 +3463,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