WIP: Improve MariaDB support

.github/workflows/test_and_publish.yml

@@ -54,6 +54,22 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
+      mariadb:
+        image: mariadb:latest
+        ports:
+          - 3307
+        env:
+          MARIADB_USER: gis
+          MARIADB_PASSWORD: gis
+          MARIADB_DATABASE: gis
+          MARIADB_ROOT_PASSWORD: gis
+        # Set health checks to wait until MariaDB has started
+        options: >-
+          --health-cmd="healthcheck.sh --connect --innodb_initialized"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=3
+
 
     steps:
 
@@ -123,6 +139,7 @@ jobs:
         PROJ_LIB: /home/runner/micromamba/envs/test_${{ matrix.python-version.flag }}/share/proj
         COVERAGE_FILE: .coverage
         PYTEST_MYSQL_DB_URL: mysql://gis:[email protected]/gis
+        PYTEST_MARIADB_DB_URL: mariadb://gis:[email protected]:3307/gis
       run: |
         # Run the unit test suite with SQLAlchemy=1.4.* and then with the latest version of SQLAlchemy
         tox -vv

.pre-commit-config.yaml

@@ -1,5 +1,3 @@
-default_language_version:
-  python: python3.8
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.6.0

geoalchemy2/admin/dialects/mysql.py

@@ -31,11 +31,16 @@ def reflect_geometry_column(inspector, table, column_info):
     column_name = column_info.get("name")
     schema = table.schema or inspector.default_schema_name
 
+    if inspector.dialect.name == "mariadb":
+        select_srid = "-1, "
+    else:
+        select_srid = "SRS_ID, "
+
     # Check geometry type, SRID and if the column is nullable
-    geometry_type_query = """SELECT DATA_TYPE, SRS_ID, IS_NULLABLE
+    geometry_type_query = """SELECT DATA_TYPE, {}IS_NULLABLE
         FROM INFORMATION_SCHEMA.COLUMNS
         WHERE TABLE_NAME = '{}' and COLUMN_NAME = '{}'""".format(
-        table.name, column_name
+        select_srid, table.name, column_name
     )
     if schema is not None:
         geometry_type_query += """ and table_schema = '{}'""".format(schema)

geoalchemy2/functions.py

@@ -254,6 +254,11 @@ def __init__(self, *args, **kwargs) -> None:
                 else:
                     func_name = elem.geom_from
                     func_args = [elem.data, elem.srid]
+                print(
+                    "$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$ IN functions.GenericFunction",
+                    elem,
+                    elem.data,
+                )
                 args_list[idx] = getattr(functions.func, func_name)(*func_args)
         _GeoFunctionParent.__init__(self, *args_list, **kwargs)
 

geoalchemy2/types/__init__.py

@@ -174,6 +174,7 @@ def bind_processor(self, dialect):
         """Specific bind_processor that automatically process spatial elements."""
 
         def process(bindvalue):
+            print("=================================================", type(bindvalue), bindvalue)
             return select_dialect(dialect.name).bind_processor_process(self, bindvalue)
 
         return process
@@ -198,17 +199,17 @@ def check_ctor_args(geometry_type, srid, dimension, use_typmod, nullable):
         return geometry_type, srid
 
 
-@compiles(_GISType, "mariadb")
 @compiles(_GISType, "mysql")
-def get_col_spec(self, *args, **kwargs):
+@compiles(_GISType, "mariadb")
+def get_col_spec_mysql(self, compiler, *args, **kwargs):
     if self.geometry_type is not None:
         spec = "%s" % self.geometry_type
     else:
         spec = "GEOMETRY"
 
     if not self.nullable or self.spatial_index:
         spec += " NOT NULL"
-    if self.srid > 0:
+    if self.srid > 0 and compiler.dialect.name != "mariadb":
         spec += " SRID %d" % self.srid
     return spec
 

test_container/helpers/init_mysql.sh

@@ -6,9 +6,10 @@ if [ $(whoami) != "root" ]; then
     exit 1
 fi
 
+echo "Starting mysql server"
 /etc/init.d/mysql start
 
-echo "waiting for mysql to start"
+echo "Waiting for mysql to start"
 while ! mysqladmin ping -h 127.0.0.1 --silent; do
     sleep 0.2
 done

tests/conftest.py

@@ -4,8 +4,10 @@
 import pytest
 from sqlalchemy import MetaData
 from sqlalchemy import create_engine
+from sqlalchemy import text
 from sqlalchemy.dialects.mysql.base import MySQLDialect
 from sqlalchemy.dialects.sqlite.base import SQLiteDialect
+from sqlalchemy.exc import InvalidRequestError
 from sqlalchemy.orm import declarative_base
 from sqlalchemy.orm import sessionmaker
 
@@ -42,7 +44,12 @@ def pytest_addoption(parser):
     parser.addoption(
         "--mysql_dburl",
         action="store",
-        help="MySQL DB URL used for tests with MySQL (`mysql://user:password@host:port/dbname`).",
+        help="MySQL DB URL used for tests (`mysql://user:password@host:port/dbname`).",
+    )
+    parser.addoption(
+        "--mariadb_dburl",
+        action="store",
+        help="MariaDB DB URL used for tests (`mariadb://user:password@host:port/dbname`).",
     )
     parser.addoption(
         "--engine-echo",
@@ -62,7 +69,7 @@ def pytest_generate_tests(metafunc):
         elif metafunc.module.__name__ == "tests.test_functional_sqlite":
             dialects = sqlite_dialects
         elif metafunc.module.__name__ == "tests.test_functional_mysql":
-            dialects = ["mysql"]
+            dialects = ["mysql", "mariadb"]
         elif metafunc.module.__name__ == "tests.test_functional_geopackage":
             dialects = ["geopackage"]
 
@@ -72,7 +79,7 @@ def pytest_generate_tests(metafunc):
             dialects = metafunc.cls.tested_dialects
 
         if dialects is None:
-            dialects = ["mysql", "postgresql"] + sqlite_dialects
+            dialects = ["mysql", "mariadb", "postgresql"] + sqlite_dialects
 
         if "sqlite" in dialects:
             # Order dialects
@@ -99,6 +106,15 @@ def db_url_mysql(request, tmpdir_factory):
     )
 
 
+@pytest.fixture(scope="session")
+def db_url_mariadb(request, tmpdir_factory):
+    return (
+        request.config.getoption("--mariadb_dburl")
+        or os.getenv("PYTEST_MARIADB_DB_URL")
+        or "mariadb://gis:gis@localhost/gis"
+    )
+
+
 @pytest.fixture(scope="session")
 def db_url_sqlite_spatialite3(request, tmpdir_factory):
     return (
@@ -134,16 +150,19 @@ def db_url(
     db_url_sqlite_spatialite4,
     db_url_geopackage,
     db_url_mysql,
+    db_url_mariadb,
 ):
     if request.param == "postgresql":
         return db_url_postgresql
     if request.param == "mysql":
         return db_url_mysql
-    elif request.param == "sqlite-spatialite3":
+    if request.param == "mariadb":
+        return db_url_mariadb
+    if request.param == "sqlite-spatialite3":
         return db_url_sqlite_spatialite3
-    elif request.param == "sqlite-spatialite4":
+    if request.param == "sqlite-spatialite4":
         return db_url_sqlite_spatialite4
-    elif request.param == "geopackage":
+    if request.param == "geopackage":
         return db_url_geopackage
     return None
 
@@ -157,24 +176,41 @@ def _engine_echo(request):
 @pytest.fixture
 def engine(tmpdir, db_url, _engine_echo):
     """Provide an engine to test database."""
-    if db_url.startswith("sqlite:///"):
-        # Copy the input SQLite DB to a temporary file and return an engine to it
-        input_url = str(db_url)[10:]
-        output_file = "test_spatial_db.sqlite"
-        current_engine = copy_and_connect_sqlite_db(
-            input_url, tmpdir / output_file, _engine_echo, "sqlite"
-        )
-    elif db_url.startswith("gpkg:///"):
-        # Copy the input SQLite DB to a temporary file and return an engine to it
-        input_url = str(db_url)[8:]
-        output_file = "test_spatial_db.gpkg"
-        current_engine = copy_and_connect_sqlite_db(
-            input_url, tmpdir / output_file, _engine_echo, "gpkg"
-        )
-    else:
-        # For other dialects the engine is directly returned
-        current_engine = create_engine(db_url, echo=_engine_echo)
-        current_engine.update_execution_options(search_path=["gis", "public"])
+    try:
+        if db_url.startswith("sqlite:///"):
+            # Copy the input SQLite DB to a temporary file and return an engine to it
+            input_url = str(db_url)[10:]
+            output_file = "test_spatial_db.sqlite"
+            current_engine = copy_and_connect_sqlite_db(
+                input_url, tmpdir / output_file, _engine_echo, "sqlite"
+            )
+        elif db_url.startswith("gpkg:///"):
+            # Copy the input GeoPackage to a temporary file and return an engine to it
+            input_url = str(db_url)[8:]
+            output_file = "test_spatial_db.gpkg"
+            current_engine = copy_and_connect_sqlite_db(
+                input_url, tmpdir / output_file, _engine_echo, "gpkg"
+            )
+        else:
+            # For other dialects the engine is directly returned
+            current_engine = create_engine(db_url, echo=_engine_echo)
+            current_engine.update_execution_options(search_path=["gis", "public"])
+    except Exception:
+        pytest.skip(reason=f"Could not create engine for this URL: {db_url}")
+
+    # Disambiguate MySQL and MariaDB
+    if current_engine.dialect.name in ["mysql", "mariadb"]:
+        try:
+            with current_engine.begin() as connection:
+                mysql_type = (
+                    "MariaDB"
+                    if "mariadb" in connection.execute(text("SELECT VERSION();")).scalar().lower()
+                    else "MySQL"
+                )
+            if current_engine.dialect.name != mysql_type.lower():
+                pytest.skip(reason=f"Can not execute {mysql_type} queries on {db_url}")
+        except InvalidRequestError:
+            pytest.skip(reason=f"Can not execute MariaDB queries on {db_url}")
 
     yield current_engine
     current_engine.dispose()

tests/schema_fixtures.py

@@ -189,7 +189,7 @@ class Lake(base):
             geom_no_idx = Column(
                 Geometry(geometry_type="LINESTRING", srid=4326, spatial_index=False)
             )
-            if dialect_name != "mysql":
+            if dialect_name not in ["mysql", "mariadb"]:
                 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))

tests/test_functional.py

@@ -249,7 +249,11 @@ def test_insert(self, conn, Lake, setup_tables):
 
         row = rows[0]
         assert isinstance(row[1], WKBElement)
+        # import pdb
+        # pdb.set_trace()
+        print("++++++++++++++++++++++++++++++++++++++++++++++++")
         wkt = conn.execute(row[1].ST_AsText()).scalar()
+        print("++++++++++++++++++++++++++++++++++++++++++++++++")
         assert format_wkt(wkt) == "LINESTRING(0 0,1 1)"
         srid = conn.execute(row[1].ST_SRID()).scalar()
         assert srid == 4326

tests/test_functional_mysql.py

@@ -42,7 +42,8 @@ def test_create_drop_tables(
 
 class TestInsertionCore:
     @pytest.mark.parametrize("use_executemany", [True, False])
-    def test_insert(self, conn, Lake, setup_tables, use_executemany):
+    @test_only_with_dialects("mysql")
+    def test_insert_mysql(self, conn, Lake, setup_tables, use_executemany):
         # Issue several inserts using DBAPI's executemany() method or single inserts. This tests
         # the Geometry type's bind_processor and bind_expression functions.
         elements = [
@@ -98,6 +99,48 @@ def test_insert(self, conn, Lake, setup_tables, use_executemany):
             [{"geom": row[1]} for row in rows],
         )
 
+    @pytest.mark.parametrize("use_executemany", [True, False])
+    @test_only_with_dialects("mariadb")
+    def test_insert_mariadb(self, conn, Lake, setup_tables, use_executemany):
+        # Issue several inserts using DBAPI's executemany() method or single inserts. This tests
+        # the Geometry type's bind_processor and bind_expression functions.
+        elements = [
+            {"geom": "LINESTRING(0 0,1 1)"},
+            {"geom": WKTElement("LINESTRING(0 0,2 2)")},
+        ]
+
+        if use_executemany:
+            conn.execute(Lake.__table__.insert(), elements)
+        else:
+            for element in elements:
+                query = Lake.__table__.insert().values(**element)
+                conn.execute(query)
+
+        results = conn.execute(Lake.__table__.select().order_by("id"))
+        rows = results.fetchall()
+
+        row = rows[0]
+        assert isinstance(row[1], WKBElement)
+        wkt = conn.execute(row[1].ST_AsText()).scalar()
+        assert wkt == "LINESTRING(0 0,1 1)"
+        srid = conn.execute(row[1].ST_SRID()).scalar()
+        assert srid == -1
+
+        row = rows[1]
+        assert isinstance(row[1], WKBElement)
+        wkt = conn.execute(row[1].ST_AsText()).scalar()
+        assert wkt == "LINESTRING(0 0,2 2)"
+        srid = conn.execute(row[1].ST_SRID()).scalar()
+        assert srid == -1
+
+        # Check that selected elements can be inserted again
+        for row in rows:
+            conn.execute(Lake.__table__.insert().values(geom=row[1]))
+        conn.execute(
+            Lake.__table__.insert(),
+            [{"geom": row[1]} for row in rows],
+        )
+
 
 class TestInsertionORM:
     def test_WKT(self, session, Lake, setup_tables):
@@ -327,13 +370,13 @@ def test_insert(self, conn, Lake, setup_tables):
 
 class TestReflection:
     @pytest.fixture
-    def create_temp_db(self, request, conn, reflection_tables_metadata):
+    def create_temp_db(self, request, conn, reflection_tables_metadata, dialect_name):
         """Temporary database, that is dropped on fixture teardown.
         Used to make sure reflection methods always uses the correct schema.
         """
         temp_db_name = "geoalchemy_test_reflection"
         engine = create_engine(
-            f"mysql://gis:gis@localhost/{temp_db_name}",
+            f"{dialect_name}://gis:gis@localhost/{temp_db_name}",
             echo=request.config.getoption("--engine-echo"),
         )
         conn.execute(text(f"CREATE DATABASE IF NOT EXISTS {temp_db_name};"))
@@ -349,19 +392,19 @@ 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, create_temp_db):
+    def test_reflection_mysql(self, conn, setup_reflection_tables, create_temp_db, dialect_name):
         t = Table("lake", MetaData(), autoload_with=conn)
 
         type_ = t.c.geom.type
         assert isinstance(type_, Geometry)
         assert type_.geometry_type == "LINESTRING"
-        assert type_.srid == 4326
+        assert type_.srid == 4326 if dialect_name == "mysql" else -1
         assert type_.dimension == 2
 
         type_ = t.c.geom_no_idx.type
         assert isinstance(type_, Geometry)
         assert type_.geometry_type == "LINESTRING"
-        assert type_.srid == 4326
+        assert type_.srid == 4326 if dialect_name == "mysql" else -1
         assert type_.dimension == 2
 
         # Drop the table