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]]:
@@ -1662,8 +1662,14 @@ def execute(self, sql: str | Select | TextClause, params=None):
         """Simple passthrough to SQLAlchemy connectable"""
         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)
+            try:
+                return self.con.exec_driver_sql(sql, *args)
+            except Exception as exc:
+                raise DatabaseError("foo") from exc
+        try:
+            return self.con.execute(sql, *args)
+        except Exception as exc:
+            raise DatabaseError from exc
 
     def read_table(
         self,
@@ -2077,9 +2083,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()
 
@@ -2148,6 +2154,8 @@ def execute(self, sql: str | Select | TextClause, params=None):
 
             ex = DatabaseError(f"Execution failed on sql '{sql}': {exc}")
             raise ex from exc
+        finally:
+            cur.close()
 
     def read_table(
         self,
@@ -2403,9 +2411,12 @@ def to_sql(
             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 Exception as exc:
+                raise DatabaseError("foo") from exc
 
         self.con.commit()
         return total_inserted
@@ -2431,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}")
 
     def _create_sql_schema(
         self,
@@ -2553,7 +2563,10 @@ def insert_statement(self, *, num_rows: int) -> str:
 
     def _execute_insert(self, conn, keys, data_iter) -> int:
         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 Exception as exc:
+            raise DatabaseError("foo") from exc
         return conn.rowcount
 
     def _execute_insert_multi(self, conn, keys, data_iter) -> int:

pandas/tests/io/test_sql.py

@@ -605,7 +605,7 @@ def mysql_pymysql_engine():
     sqlalchemy = pytest.importorskip("sqlalchemy")
     pymysql = pytest.importorskip("pymysql")
     engine = sqlalchemy.create_engine(
-        "mysql+pymysql://root@localhost:3306/pandas",
+        "mysql+pymysql://root@mysql:3306/pandas",
         connect_args={"client_flag": pymysql.constants.CLIENT.MULTI_STATEMENTS},
         poolclass=sqlalchemy.pool.NullPool,
     )
@@ -653,7 +653,7 @@ def postgresql_psycopg2_engine():
     sqlalchemy = pytest.importorskip("sqlalchemy")
     pytest.importorskip("psycopg2")
     engine = sqlalchemy.create_engine(
-        "postgresql+psycopg2://postgres:postgres@localhost:5432/pandas",
+        "postgresql+psycopg2://postgres:postgres@postgres:5432/pandas",
         poolclass=sqlalchemy.pool.NullPool,
     )
     yield engine
@@ -689,7 +689,7 @@ def postgresql_adbc_conn():
     pytest.importorskip("adbc_driver_postgresql")
     from adbc_driver_postgresql import dbapi
 
-    uri = "postgresql://postgres:postgres@localhost:5432/pandas"
+    uri = "postgresql://postgres:postgres@postgres:5432/pandas"
     with dbapi.connect(uri) as conn:
         yield conn
         for view in get_all_views(conn):
@@ -2595,7 +2595,7 @@ def test_sql_open_close(test_frame3):
 
 @td.skip_if_installed("sqlalchemy")
 def test_con_string_import_error():
-    conn = "mysql://root@localhost/pandas"
+    conn = "mysql://root@mysql/pandas"
     msg = "Using URI string without sqlalchemy installed"
     with pytest.raises(ImportError, match=msg):
         sql.read_sql("SELECT * FROM iris", conn)
@@ -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