Fix sqlalchemy 1.4.46 warning (#50558)

Author: Chuck Cadman

pandas/io/sql.py

@@ -319,7 +319,7 @@ def read_sql_table(
     --------
     >>> pd.read_sql_table('table_name', 'postgres:///db_name')  # doctest:+SKIP
     """
-    with pandasSQL_builder(con, schema=schema) as pandas_sql:
+    with pandasSQL_builder(con, schema=schema, need_transaction=True) as pandas_sql:
         if not pandas_sql.has_table(table_name):
             raise ValueError(f"Table {table_name} not found")
 
@@ -883,7 +883,8 @@ def sql_schema(self) -> str:
     def _execute_create(self) -> None:
         # Inserting table into database, add to MetaData object
         self.table = self.table.to_metadata(self.pd_sql.meta)
-        self.table.create(bind=self.pd_sql.con)
+        with self.pd_sql.run_transaction():
+            self.table.create(bind=self.pd_sql.con)
 
     def create(self) -> None:
         if self.exists():
@@ -1529,7 +1530,11 @@ def __exit__(self, *args) -> None:
 
     @contextmanager
     def run_transaction(self):
-        yield self.con
+        if not self.con.in_transaction():
+            with self.con.begin():
+                yield self.con
+        else:
+            yield self.con
 
     def execute(self, sql: str | Select | TextClause, params=None):
         """Simple passthrough to SQLAlchemy connectable"""
@@ -1932,7 +1937,8 @@ def drop_table(self, table_name: str, schema: str | None = None) -> None:
         schema = schema or self.meta.schema
         if self.has_table(table_name, schema):
             self.meta.reflect(bind=self.con, only=[table_name], schema=schema)
-            self.get_table(table_name, schema).drop(bind=self.con)
+            with self.run_transaction():
+                self.get_table(table_name, schema).drop(bind=self.con)
             self.meta.clear()
 
     def _create_sql_schema(

pandas/tests/io/test_sql.py

@@ -149,8 +149,6 @@ def create_and_load_iris(conn, iris_file: Path, dialect: str):
     from sqlalchemy.engine import Engine
 
     iris = iris_table_metadata(dialect)
-    iris.drop(conn, checkfirst=True)
-    iris.create(bind=conn)
 
     with iris_file.open(newline=None) as csvfile:
         reader = csv.reader(csvfile)
@@ -160,9 +158,14 @@ def create_and_load_iris(conn, iris_file: Path, dialect: str):
         if isinstance(conn, Engine):
             with conn.connect() as conn:
                 with conn.begin():
+                    iris.drop(conn, checkfirst=True)
+                    iris.create(bind=conn)
                     conn.execute(stmt)
         else:
-            conn.execute(stmt)
+            with conn.begin():
+                iris.drop(conn, checkfirst=True)
+                iris.create(bind=conn)
+                conn.execute(stmt)
 
 
 def create_and_load_iris_view(conn):
@@ -180,7 +183,8 @@ def create_and_load_iris_view(conn):
                 with conn.begin():
                     conn.execute(stmt)
         else:
-            conn.execute(stmt)
+            with conn.begin():
+                conn.execute(stmt)
 
 
 def types_table_metadata(dialect: str):
@@ -243,16 +247,19 @@ def create_and_load_types(conn, types_data: list[dict], dialect: str):
     from sqlalchemy.engine import Engine
 
     types = types_table_metadata(dialect)
-    types.drop(conn, checkfirst=True)
-    types.create(bind=conn)
 
     stmt = insert(types).values(types_data)
     if isinstance(conn, Engine):
         with conn.connect() as conn:
             with conn.begin():
+                types.drop(conn, checkfirst=True)
+                types.create(bind=conn)
                 conn.execute(stmt)
     else:
-        conn.execute(stmt)
+        with conn.begin():
+            types.drop(conn, checkfirst=True)
+            types.create(bind=conn)
+            conn.execute(stmt)
 
 
 def check_iris_frame(frame: DataFrame):
@@ -713,7 +720,8 @@ def test_read_procedure(conn, request):
             with engine_conn.begin():
                 engine_conn.execute(proc)
     else:
-        conn.execute(proc)
+        with conn.begin():
+            conn.execute(proc)
 
     res1 = sql.read_sql_query("CALL get_testdb();", conn)
     tm.assert_frame_equal(df, res1)
@@ -995,8 +1003,6 @@ class _TestSQLApi(PandasSQLTest):
     @pytest.fixture(autouse=True)
     def setup_method(self, iris_path, types_data):
         self.conn = self.connect()
-        if not isinstance(self.conn, sqlite3.Connection):
-            self.conn.begin()
         self.load_iris_data(iris_path)
         self.load_types_data(types_data)
         self.load_test_data_and_sql()
@@ -1492,7 +1498,8 @@ def test_not_reflect_all_tables(self):
                     with conn.begin():
                         conn.execute(query)
             else:
-                self.conn.execute(query)
+                with self.conn.begin():
+                    self.conn.execute(query)
 
         with tm.assert_produces_warning(None):
             sql.read_sql_table("other_table", self.conn)
@@ -1742,7 +1749,6 @@ def setup_class(cls):
     def setup_method(self, iris_path, types_data):
         try:
             self.conn = self.engine.connect()
-            self.conn.begin()
             self.pandasSQL = sql.SQLDatabase(self.conn)
         except sqlalchemy.exc.OperationalError:
             pytest.skip(f"Can't connect to {self.flavor} server")
@@ -2146,7 +2152,6 @@ def _get_index_columns(self, tbl_name):
     def test_to_sql_save_index(self):
         self._to_sql_save_index()
 
-    @pytest.mark.xfail(reason="Nested transactions rollbacks don't work with Pandas")
     def test_transactions(self):
         self._transaction_test()
 
@@ -2168,7 +2173,8 @@ def test_get_schema_create_table(self, test_frame3):
                 with conn.begin():
                     conn.execute(create_sql)
         else:
-            self.conn.execute(create_sql)
+            with self.conn.begin():
+                self.conn.execute(create_sql)
         returned_df = sql.read_sql_table(tbl, self.conn)
         tm.assert_frame_equal(returned_df, blank_test_df, check_index_type=False)
         self.drop_table(tbl, self.conn)
@@ -2578,7 +2584,8 @@ class Test(BaseModel):
             id = Column(Integer, primary_key=True)
             string_column = Column(String(50))
 
-        BaseModel.metadata.create_all(self.conn)
+        with self.conn.begin():
+            BaseModel.metadata.create_all(self.conn)
         Session = sessionmaker(bind=self.conn)
         with Session() as session:
             df = DataFrame({"id": [0, 1], "string_column": ["hello", "world"]})
@@ -2660,8 +2667,9 @@ def test_schema_support(self):
         df = DataFrame({"col1": [1, 2], "col2": [0.1, 0.2], "col3": ["a", "n"]})
 
         # create a schema
-        self.conn.exec_driver_sql("DROP SCHEMA IF EXISTS other CASCADE;")
-        self.conn.exec_driver_sql("CREATE SCHEMA other;")
+        with self.conn.begin():
+            self.conn.exec_driver_sql("DROP SCHEMA IF EXISTS other CASCADE;")
+            self.conn.exec_driver_sql("CREATE SCHEMA other;")
 
         # write dataframe to different schema's
         assert df.to_sql("test_schema_public", self.conn, index=False) == 2
@@ -2693,8 +2701,9 @@ def test_schema_support(self):
         # different if_exists options
 
         # create a schema
-        self.conn.exec_driver_sql("DROP SCHEMA IF EXISTS other CASCADE;")
-        self.conn.exec_driver_sql("CREATE SCHEMA other;")
+        with self.conn.begin():
+            self.conn.exec_driver_sql("DROP SCHEMA IF EXISTS other CASCADE;")
+            self.conn.exec_driver_sql("CREATE SCHEMA other;")
 
         # write dataframe with different if_exists options
         assert (