Slqalchemy 2 compatibility (#2846)

Co-authored-by: Rietesh <rietesh4535@gmail.com>

.github/workflows/tests.yaml

@@ -107,7 +107,7 @@ jobs:
           pytest -v -r a -n auto --color=yes --cov=geopandas --cov-append --cov-report term-missing --cov-report xml geopandas/
 
       - name: Test with PostGIS
-        if: contains(matrix.env, '39-pd13-conda-forge.yaml') && contains(matrix.os, 'ubuntu')
+        if: (contains(matrix.env, '39-pd13-conda-forge.yaml') || contains(matrix.env, '311-latest-conda-forge.yaml')) && contains(matrix.os, 'ubuntu')
         env:
           PGUSER: postgres
           PGPASSWORD: postgres

ci/envs/311-latest-conda-forge.yaml

@@ -4,7 +4,7 @@ channels:
 dependencies:
   - python=3.11
   # required
-  - pandas
+  # - pandas
   - shapely
   - fiona
   - pyproj
@@ -26,10 +26,13 @@ dependencies:
   - geopy
   # installed in tests.yaml, because not available on windows
   # - postgis
-  - SQLalchemy<2
+  - SQLalchemy>=2
   - psycopg2
   - libspatialite
   - geoalchemy2
   - pyarrow
   # doctest testing
   - pytest-doctestplus
+  - pip
+  - pip:
+    - pandas==2.0.0rc1

geopandas/io/sql.py

@@ -31,7 +31,10 @@ def _get_conn(conn_or_engine):
     from sqlalchemy.engine.base import Engine, Connection
 
     if isinstance(conn_or_engine, Connection):
-        with conn_or_engine.begin():
+        if not conn_or_engine.in_transaction():
+            with conn_or_engine.begin():
+                yield conn_or_engine
+        else:
             yield conn_or_engine
     elif isinstance(conn_or_engine, Engine):
         with conn_or_engine.begin() as conn:
@@ -395,6 +398,7 @@ def _write_postgis(
     """
     try:
         from geoalchemy2 import Geometry
+        from sqlalchemy import text
     except ImportError:
         raise ImportError("'to_postgis()' requires geoalchemy2 package.")
 
@@ -431,8 +435,10 @@ def _write_postgis(
             # Only check SRID if table exists
             if connection.dialect.has_table(connection, name, schema):
                 target_srid = connection.execute(
-                    "SELECT Find_SRID('{schema}', '{table}', '{geom_col}');".format(
-                        schema=schema_name, table=name, geom_col=geom_name
+                    text(
+                        "SELECT Find_SRID('{schema}', '{table}', '{geom_col}');".format(
+                            schema=schema_name, table=name, geom_col=geom_name
+                        )
                     )
                 ).fetchone()[0]
 

geopandas/io/tests/test_sql.py

@@ -5,6 +5,7 @@ configuration. postGIS tests require a test database to have been setup;
 see geopandas.tests.util for more information.
 """
 import os
+import warnings
 
 import pandas as pd
 
@@ -15,6 +16,13 @@ from geopandas.io.sql import _get_conn as get_conn, _write_postgis as write_post
 from geopandas.tests.util import create_postgis, create_spatialite, validate_boro_df
 import pytest
 
+try:
+    from sqlalchemy import text
+except ImportError:
+    # Avoid local imports for text in all sqlalchemy tests
+    # all tests using text use engine_postgis, which ensures sqlalchemy is available
+    text = str
+
 
 @pytest.fixture
 def df_nybb():
@@ -43,8 +51,11 @@ def connection_postgis():
         )
     except OperationalError:
         pytest.skip("Cannot connect with postgresql database")
-
-    yield con
+    with warnings.catch_warnings():
+        warnings.filterwarnings(
+            "ignore", message="pandas only supports SQLAlchemy connectable.*"
+        )
+        yield con
     con.close()
 
 
@@ -73,7 +84,7 @@ def engine_postgis():
                 port=port,
             )
         )
-        con.begin()
+        con.connect()
     except Exception:
         pytest.skip("Cannot connect with postgresql database")
 
@@ -115,11 +126,15 @@ def drop_table_if_exists(conn_or_engine, table):
     sqlalchemy = pytest.importorskip("sqlalchemy")
 
     if sqlalchemy.inspect(conn_or_engine).has_table(table):
-        metadata = sqlalchemy.MetaData(conn_or_engine)
-        metadata.reflect()
+        metadata = sqlalchemy.MetaData()
+        with warnings.catch_warnings():
+            warnings.filterwarnings(
+                "ignore", message="Did not recognize type 'geometry' of column.*"
+            )
+            metadata.reflect(conn_or_engine)
         table = metadata.tables.get(table)
         if table is not None:
-            table.drop(checkfirst=True)
+            table.drop(conn_or_engine, checkfirst=True)
 
 
 @pytest.fixture
@@ -340,7 +355,7 @@ class TestIO:
         # Write to db
         write_postgis(df_nybb, con=engine, name=table, if_exists="fail")
         # Validate
-        sql = "SELECT * FROM {table};".format(table=table)
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         validate_boro_df(df)
 
@@ -355,7 +370,7 @@ class TestIO:
         # Write to db
         write_postgis(df_nybb, con=engine, name=table, if_exists="fail")
         # Validate
-        sql = 'SELECT * FROM "{table}";'.format(table=table)
+        sql = text('SELECT * FROM "{table}";'.format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         validate_boro_df(df)
 
@@ -370,7 +385,7 @@ class TestIO:
             # Write to db
             write_postgis(df_nybb, con=con, name=table, if_exists="fail")
             # Validate
-            sql = "SELECT * FROM {table};".format(table=table)
+            sql = text("SELECT * FROM {table};".format(table=table))
             df = read_postgis(sql, con, geom_col="geometry")
             validate_boro_df(df)
 
@@ -406,7 +421,7 @@ class TestIO:
         # Overwrite
         write_postgis(df_nybb, con=engine, name=table, if_exists="replace")
         # Validate
-        sql = "SELECT * FROM {table};".format(table=table)
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         validate_boro_df(df)
 
@@ -423,7 +438,7 @@ class TestIO:
         write_postgis(df_nybb, con=engine, name=table, if_exists="replace")
         write_postgis(df_nybb, con=engine, name=table, if_exists="append")
         # Validate
-        sql = "SELECT * FROM {table};".format(table=table)
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         new_rows, new_cols = df.shape
 
@@ -449,13 +464,16 @@ class TestIO:
         # Write to db
         df_nybb = df_nybb
         df_nybb.crs = None
-        write_postgis(df_nybb, con=engine, name=table, if_exists="replace")
+        with pytest.warns(UserWarning, match="Could not parse CRS from the GeoDataF"):
+            write_postgis(df_nybb, con=engine, name=table, if_exists="replace")
         # Validate that srid is -1
-        target_srid = engine.execute(
+        sql = text(
             "SELECT Find_SRID('{schema}', '{table}', '{geom_col}');".format(
                 schema="public", table=table, geom_col="geometry"
             )
-        ).fetchone()[0]
+        )
+        with engine.connect() as conn:
+            target_srid = conn.execute(sql).fetchone()[0]
         assert target_srid == 0, "SRID should be 0, found %s" % target_srid
 
     def test_write_postgis_with_esri_authority(self, engine_postgis, df_nybb):
@@ -471,11 +489,13 @@ class TestIO:
         df_nybb_esri = df_nybb.to_crs("ESRI:102003")
         write_postgis(df_nybb_esri, con=engine, name=table, if_exists="replace")
         # Validate that srid is 102003
-        target_srid = engine.execute(
+        sql = text(
             "SELECT Find_SRID('{schema}', '{table}', '{geom_col}');".format(
                 schema="public", table=table, geom_col="geometry"
             )
-        ).fetchone()[0]
+        )
+        with engine.connect() as conn:
+            target_srid = conn.execute(sql).fetchone()[0]
         assert target_srid == 102003, "SRID should be 102003, found %s" % target_srid
 
     def test_write_postgis_geometry_collection(
@@ -491,11 +511,14 @@ class TestIO:
         write_postgis(df_geom_collection, con=engine, name=table, if_exists="replace")
 
         # Validate geometry type
-        sql = "SELECT DISTINCT(GeometryType(geometry)) FROM {table} ORDER BY 1;".format(
-            table=table
+        sql = text(
+            "SELECT DISTINCT(GeometryType(geometry)) FROM {table} ORDER BY 1;".format(
+                table=table
+            )
         )
-        geom_type = engine.execute(sql).fetchone()[0]
-        sql = "SELECT * FROM {table};".format(table=table)
+        with engine.connect() as conn:
+            geom_type = conn.execute(sql).fetchone()[0]
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
 
         assert geom_type.upper() == "GEOMETRYCOLLECTION"
@@ -516,10 +539,13 @@ class TestIO:
         )
 
         # Validate geometry type
-        sql = "SELECT DISTINCT GeometryType(geometry) FROM {table} ORDER BY 1;".format(
-            table=table
+        sql = text(
+            "SELECT DISTINCT GeometryType(geometry) FROM {table} ORDER BY 1;".format(
+                table=table
+            )
         )
-        res = engine.execute(sql).fetchall()
+        with engine.connect() as conn:
+            res = conn.execute(sql).fetchall()
         assert res[0][0].upper() == "LINESTRING"
         assert res[1][0].upper() == "MULTILINESTRING"
         assert res[2][0].upper() == "POINT"
@@ -535,10 +561,13 @@ class TestIO:
         write_postgis(df_linear_ring, con=engine, name=table, if_exists="replace")
 
         # Validate geometry type
-        sql = "SELECT DISTINCT(GeometryType(geometry)) FROM {table} ORDER BY 1;".format(
-            table=table
+        sql = text(
+            "SELECT DISTINCT(GeometryType(geometry)) FROM {table} ORDER BY 1;".format(
+                table=table
+            )
         )
-        geom_type = engine.execute(sql).fetchone()[0]
+        with engine.connect() as conn:
+            geom_type = conn.execute(sql).fetchone()[0]
 
         assert geom_type.upper() == "LINESTRING"
 
@@ -558,15 +587,19 @@ class TestIO:
             chunksize=1,
         )
         # Validate row count
-        sql = "SELECT COUNT(geometry) FROM {table};".format(table=table)
-        row_cnt = engine.execute(sql).fetchone()[0]
+        sql = text("SELECT COUNT(geometry) FROM {table};".format(table=table))
+        with engine.connect() as conn:
+            row_cnt = conn.execute(sql).fetchone()[0]
         assert row_cnt == 3
 
         # Validate geometry type
-        sql = "SELECT DISTINCT GeometryType(geometry) FROM {table} ORDER BY 1;".format(
-            table=table
+        sql = text(
+            "SELECT DISTINCT GeometryType(geometry) FROM {table} ORDER BY 1;".format(
+                table=table
+            )
         )
-        res = engine.execute(sql).fetchall()
+        with engine.connect() as conn:
+            res = conn.execute(sql).fetchall()
         assert res[0][0].upper() == "LINESTRING"
         assert res[1][0].upper() == "MULTILINESTRING"
         assert res[2][0].upper() == "POINT"
@@ -579,15 +612,16 @@ class TestIO:
 
         table = "nybb"
         schema_to_use = "test"
-        sql = "CREATE SCHEMA IF NOT EXISTS {schema};".format(schema=schema_to_use)
-        engine.execute(sql)
+        sql = text("CREATE SCHEMA IF NOT EXISTS {schema};".format(schema=schema_to_use))
+        with engine.begin() as conn:
+            conn.execute(sql)
 
         write_postgis(
             df_nybb, con=engine, name=table, if_exists="replace", schema=schema_to_use
         )
         # Validate
-        sql = "SELECT * FROM {schema}.{table};".format(
-            schema=schema_to_use, table=table
+        sql = text(
+            "SELECT * FROM {schema}.{table};".format(schema=schema_to_use, table=table)
         )
 
         df = read_postgis(sql, engine, geom_col="geometry")
@@ -603,16 +637,19 @@ class TestIO:
 
         table = "nybb"
         schema_to_use = "test"
-        sql = "CREATE SCHEMA IF NOT EXISTS {schema};".format(schema=schema_to_use)
-        engine.execute(sql)
+        sql = text("CREATE SCHEMA IF NOT EXISTS {schema};".format(schema=schema_to_use))
+        with engine.begin() as conn:
+            conn.execute(sql)
 
         try:
             write_postgis(
                 df_nybb, con=engine, name=table, if_exists="fail", schema=schema_to_use
             )
             # Validate
-            sql = "SELECT * FROM {schema}.{table};".format(
-                schema=schema_to_use, table=table
+            sql = text(
+                "SELECT * FROM {schema}.{table};".format(
+                    schema=schema_to_use, table=table
+                )
             )
 
             df = read_postgis(sql, engine, geom_col="geometry")
@@ -627,8 +664,8 @@ class TestIO:
             df_nybb, con=engine, name=table, if_exists="replace", schema=schema_to_use
         )
         # Validate
-        sql = "SELECT * FROM {schema}.{table};".format(
-            schema=schema_to_use, table=table
+        sql = text(
+            "SELECT * FROM {schema}.{table};".format(schema=schema_to_use, table=table)
         )
 
         df = read_postgis(sql, engine, geom_col="geometry")
@@ -645,7 +682,7 @@ class TestIO:
         write_postgis(df_3D_geoms, con=engine, name=table, if_exists="replace")
 
         # Check that all geometries have 3 dimensions
-        sql = "SELECT * FROM {table};".format(table=table)
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         assert list(df.geometry.has_z) == [True, True, True]
 
@@ -661,7 +698,7 @@ class TestIO:
         write_postgis(df_nybb, con=engine, name=table, if_exists="replace")
 
         # Check that the row order matches
-        sql = "SELECT * FROM {table};".format(table=table)
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         assert df["BoroCode"].tolist() == correct_order
 
@@ -678,7 +715,7 @@ class TestIO:
         write_postgis(df_nybb, con=engine, name=table, if_exists="append")
 
         # Check that the row order matches
-        sql = "SELECT * FROM {table};".format(table=table)
+        sql = text("SELECT * FROM {table};".format(table=table))
         df = read_postgis(sql, engine, geom_col="geometry")
         validate_boro_df(df)