fix: handle mysql schemas in column reflection

geoalchemy2/admin/dialects/mysql.py

@@ -29,15 +29,16 @@ def reflect_geometry_column(inspector, table, column_info):
         return
 
     column_name = column_info.get("name")
+    schema = table.schema or inspector.default_schema_name
 
     # Check geometry type, SRID and if the column is nullable
     geometry_type_query = """SELECT DATA_TYPE, SRS_ID, IS_NULLABLE
         FROM INFORMATION_SCHEMA.COLUMNS
         WHERE TABLE_NAME = '{}' and COLUMN_NAME = '{}'""".format(
         table.name, column_name
     )
-    if table.schema is not None:
-        geometry_type_query += """ and table_schema = '{}'""".format(table.schema)
+    if schema is not None:
+        geometry_type_query += """ and table_schema = '{}'""".format(schema)
     geometry_type, srid, nullable_str = inspector.bind.execute(text(geometry_type_query)).one()
     is_nullable = str(nullable_str).lower() == "yes"
 
@@ -51,8 +52,8 @@ def reflect_geometry_column(inspector, table, column_info):
         WHERE TABLE_NAME = '{}' and COLUMN_NAME = '{}'""".format(
         table.name, column_name
     )
-    if table.schema is not None:
-        has_index_query += """ and TABLE_SCHEMA = '{}'""".format(table.schema)
+    if schema is not None:
+        has_index_query += """ and TABLE_SCHEMA = '{}'""".format(schema)
     spatial_index_res = inspector.bind.execute(text(has_index_query)).scalar()
     spatial_index = str(spatial_index_res).lower() == "spatial"
 

tests/schema_fixtures.py

@@ -183,7 +183,7 @@ class IndexTestWithoutSchema(base):
 
 
 @pytest.fixture
-def reflection_tables_metadata():
+def reflection_tables_metadata(engine):
     metadata = MetaData()
     base = declarative_base(metadata=metadata)
 
@@ -192,8 +192,9 @@ class Lake(base):
         id = Column(Integer, primary_key=True)
         geom = Column(Geometry(geometry_type="LINESTRING", srid=4326))
         geom_no_idx = Column(Geometry(geometry_type="LINESTRING", srid=4326, spatial_index=False))
-        geom_z = Column(Geometry(geometry_type="LINESTRINGZ", srid=4326, dimension=3))
-        geom_m = Column(Geometry(geometry_type="LINESTRINGM", srid=4326, dimension=3))
-        geom_zm = Column(Geometry(geometry_type="LINESTRINGZM", srid=4326, dimension=4))
+        if engine.dialect.name != "mysql":
+            geom_z = Column(Geometry(geometry_type="LINESTRINGZ", srid=4326, dimension=3))
+            geom_m = Column(Geometry(geometry_type="LINESTRINGM", srid=4326, dimension=3))
+            geom_zm = Column(Geometry(geometry_type="LINESTRINGZM", srid=4326, dimension=4))
 
     return metadata

tests/test_functional_mysql.py

@@ -9,7 +9,7 @@
 from sqlalchemy import Table
 from sqlalchemy import __version__ as SA_VERSION
 from sqlalchemy import bindparam
-from sqlalchemy import text
+from sqlalchemy import text, create_engine
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.exc import StatementError
 from sqlalchemy.orm import declarative_base
@@ -329,26 +329,34 @@ def test_insert(self, conn, Lake, setup_tables):
 
 class TestReflection:
     @pytest.fixture
-    def reflection_tables_metadata_mysql(self):
-        metadata = MetaData()
-        base = declarative_base(metadata=metadata)
-
-        class Lake(base):
-            __tablename__ = "lake"
-            id = Column(Integer, primary_key=True)
-            geom = Column(Geometry(geometry_type="LINESTRING", srid=4326, nullable=False))
-            geom_no_idx = Column(
-                Geometry(geometry_type="LINESTRING", srid=4326, spatial_index=False, nullable=True)
-            )
-
-        return metadata
+    def create_temp_db(self, reflection_tables_metadata):
+        """
+        Temporary database, that is dropped on fixture teardown.
+        Used to make sure reflection methods always uses the correct schema.
+        """
+        engine = create_engine("mysql://gis:gis@localhost")
+        temp_db_name = "geoalchemy_test_reflection"
+        drop = text(f"DROP DATABASE IF EXISTS {temp_db_name};")
+
+        with engine.connect() as connection:
+            with connection.begin():
+                connection.execute(drop)
+                connection.execute(text(f"CREATE DATABASE {temp_db_name};"))
+                connection.execute(text(f"USE {temp_db_name};"))
+                reflection_tables_metadata.drop_all(connection, checkfirst=True)
+                reflection_tables_metadata.create_all(connection)
+
+            yield
+
+            with connection.begin():
+                connection.execute(drop)
 
     @pytest.fixture
-    def setup_reflection_tables(self, reflection_tables_metadata_mysql, conn):
-        reflection_tables_metadata_mysql.drop_all(conn, checkfirst=True)
-        reflection_tables_metadata_mysql.create_all(conn)
+    def setup_reflection_tables(self, reflection_tables_metadata, conn):
+        reflection_tables_metadata.drop_all(conn, checkfirst=True)
+        reflection_tables_metadata.create_all(conn)
 
-    def test_reflection_mysql(self, conn, setup_reflection_tables):
+    def test_reflection_mysql(self, conn, setup_reflection_tables, create_temp_db):
         t = Table("lake", MetaData(), autoload_with=conn)
 
         type_ = t.c.geom.type
@@ -386,8 +394,11 @@ def test_reflection_mysql(self, conn, setup_reflection_tables):
                 B.INDEX_TYPE
             FROM INFORMATION_SCHEMA.COLUMNS AS A
             LEFT JOIN INFORMATION_SCHEMA.STATISTICS AS B
-            ON A.TABLE_NAME = B.TABLE_NAME AND A.COLUMN_NAME = B.COLUMN_NAME
-            WHERE A.TABLE_SCHEMA = 'gis' AND A.TABLE_NAME = 'lake'
+            ON A.TABLE_NAME = B.TABLE_NAME 
+                AND A.COLUMN_NAME = B.COLUMN_NAME
+                AND A.TABLE_SCHEMA = B.TABLE_SCHEMA
+            WHERE A.TABLE_SCHEMA = 'gis'
+            AND A.TABLE_NAME = 'lake'
             ORDER BY TABLE_NAME, COLUMN_NAME;"""
         )