rewrite geopackage conversion

.github/workflows/test.yml

@@ -44,23 +44,19 @@ jobs:
         with:
           python-version: ${{ matrix.python }}
 
-      - name: Install sqlite3 and spatialite
+      - name: Install GDAL, sqlite3 and spatialite
         run: |
           sudo apt update
-          sudo apt install --yes --no-install-recommends sqlite3 libsqlite3-mod-spatialite
+          sudo apt install --yes --no-install-recommends sqlite3 libsqlite3-mod-spatialite libgdal-dev gdal-bin
           sqlite3 --version
-
-      - name: Install gdal (for ogr2ogr)
-        run: |
-          sudo apt update          
-          sudo apt install --yes --no-install-recommends gdal-bin
           ogr2ogr --version
 
       - name: Install python dependencies
         shell: bash
         run: |
           pip install --disable-pip-version-check --upgrade pip setuptools wheel
           pip install ${{ matrix.pins }} .[test,cli]
+          pip install GDAL==$(gdal-config --version)
           pip list
 
       - name: Run tests

CHANGES.rst

@@ -6,7 +6,14 @@ Changelog of threedi-schema
 0.229.1 (unreleased)
 --------------------
 
-- Nothing changed yet.
+- Rename sqlite table "tags" to "tag"
+- Remove indices referring to removed tables in previous migrations
+- Make model_settings.use_2d_rain and model_settings.friction_averaging booleans
+- Remove columns referencing v2 in geometry_column
+- Ensure correct use_* values when matching tables have no data
+- Use custom types for comma separated and table text fields to strip extra white space
+- Correct direction of dwf and surface map
+- Remove v2 related views from sqlite
 
 
 0.229.0 (2025-01-08)

threedi_schema/application/schema.py

@@ -1,5 +1,3 @@
-import re
-import subprocess
 import warnings
 from pathlib import Path
 
@@ -10,6 +8,7 @@
 from alembic.environment import EnvironmentContext
 from alembic.migration import MigrationContext
 from alembic.script import ScriptDirectory
+from osgeo import gdal
 from sqlalchemy import Column, Integer, MetaData, Table, text
 from sqlalchemy.exc import IntegrityError
 
@@ -49,6 +48,13 @@ def _upgrade_database(db, revision="head", unsafe=True, progress_func=None):
     alembic_command.upgrade(config, revision)
 
 
+class GdalErrorHandler:
+    def __call__(self, err_level, err_no, err_msg):
+        self.err_level = err_level
+        self.err_no = err_no
+        self.err_msg = err_msg
+
+
 class ModelSchema:
     def __init__(self, threedi_db, declared_models=models.DECLARED_MODELS):
         self.db = threedi_db
@@ -205,42 +211,22 @@ def upgrade_spatialite_version(self):
 
     def convert_to_geopackage(self):
         """
-        Convert spatialite to geopackage using gdal's ogr2ogr.
+        Convert spatialite to geopackage using gdal.VectorTranslate.
 
         Does nothing if the current database is already a geopackage.
 
-        Raises UpgradeFailedError if the conversion of spatialite to geopackage with ogr2ogr fails.
+        Raises UpgradeFailedError if the conversion of spatialite to geopackage with VectorTranslate fails.
         """
+
+        handler = GdalErrorHandler()
+        gdal.PushErrorHandler(handler)
+        gdal.UseExceptions()
+
+        warnings = []
+
         if self.db.get_engine().dialect.name == "geopackage":
             return
-        # Check if ogr2ogr
-        result = subprocess.run(
-            "ogr2ogr --version",
-            shell=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            text=True,
-        )
-        ## ogr2ogr is installed; make sure the version is high enough and return if not
-        if result.returncode == 0:
-            # get version
-            version = re.findall(r"\b(\d+\.\d+\.\d+)\b", result.stdout)[0]
-            # trim patch version and convert to float
-            float_version = float(version[0 : version.rfind(".")])
-            if float_version < 3.4:
-                warnings.warn(
-                    f"ogr2ogr 3.4 (part of GDAL) or newer is needed to convert spatialite to geopackage "
-                    f"but ogr2ogr {version} was found. {self.db.path} will not be converted"
-                    f"to geopackage."
-                )
-                return
-        # ogr2ogr is not (properly) installed; return
-        elif result.returncode != 0:
-            warnings.warn(
-                f"ogr2ogr (part of GDAL) is needed to convert spatialite to geopackage but no working"
-                f"working installation was found:\n{result.stderr}"
-            )
-            return
+
         # Ensure database is upgraded and views are recreated
         self.upgrade()
         self.validate_schema()
@@ -250,46 +236,70 @@ def convert_to_geopackage(self):
             with work_db.get_session() as session:
                 session.execute(text("DROP TABLE IF EXISTS spatialite_history;"))
                 session.execute(text("DROP TABLE IF EXISTS views_geometry_columns;"))
-            cmd = [
-                "ogr2ogr",
-                "-skipfailures",
-                "-f",
-                "gpkg",
-                str(Path(self.db.path).with_suffix(".gpkg")),
-                str(work_db.path),
-                "-oo",
-                "LIST_ALL_TABLES=YES",
-            ]
-            try:
-                p = subprocess.Popen(
-                    cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=-1
+
+                all_tablenames = [model.__tablename__ for model in self.declared_models]
+                geometry_tablenames = (
+                    session.execute(text("SELECT f_table_name FROM geometry_columns;"))
+                    .scalars()
+                    .all()
+                )
+                non_geometry_tablenames = [
+                    name for name in all_tablenames if name not in geometry_tablenames
+                ]
+
+                if (
+                    session.execute(
+                        text(
+                            "SELECT count(*) FROM sqlite_master WHERE name='schema_version';"
+                        )
+                    ).scalar()
+                    > 0
+                ):
+                    non_geometry_tablenames.append("schema_version")
+
+            infile = str(work_db.path)
+            outfile = str(Path(self.db.path).with_suffix(".gpkg"))
+
+            conversion_list = []
+            conversion_list.append(
+                gdal.VectorTranslateOptions(
+                    format="gpkg",
+                    skipFailures=True,
                 )
-            except Exception as e:
-                raise UpgradeFailedError(f"ogr2ogr failed conversion:\n{e}")
-            _, out = p.communicate()
-        # Error handling
-        # convert bytes to utf and split lines
-        out_list = out.decode("utf-8").split("\n")
-        # collect only errors and remove 'ERROR #:'
-        errors = [
-            [idx, ": ".join(item.split(": ")[1:])]
-            for idx, item in enumerate(out_list)
-            if item.lower().startswith("error")
-        ]
-        # While creating the geopackage with ogr2ogr an error occurs
-        # because ogr2ogr tries to create a table `sqlite_sequence`, which
-        # is reserved for internal use. The resulting database seems fine,
-        # so this specific error is ignored
-        # convert error output to list
-        expected_error = 'sqlite3_exec(CREATE TABLE "sqlite_sequence" ( "rowid" INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, "name" TEXT, "seq" TEXT)) failed: object name reserved for internal use: sqlite_sequence'
-        unexpected_error_indices = [
-            idx for idx, error in errors if error.lower() != expected_error.lower()
-        ]
-        if len(unexpected_error_indices) > 0:
-            error_str = "\n".join(
-                [out_list[idx].decode("utf-8") for idx in unexpected_error_indices]
             )
-            raise UpgradeFailedError(f"ogr2ogr didn't finish as expected:\n{error_str}")
+            for table in non_geometry_tablenames:
+                conversion_list.append(
+                    gdal.VectorTranslateOptions(
+                        format="gpkg",
+                        accessMode="update",
+                        SQLStatement=f"SELECT * FROM {table}",
+                        layerName=table,
+                    )
+                )
+            for conversion_options in conversion_list:
+                try:
+                    ds = gdal.VectorTranslate(
+                        destNameOrDestDS=outfile,
+                        srcDS=infile,
+                        options=conversion_options,
+                    )
+                    # dereference dataset before writing additional layers to ensure the data is written
+                    del ds
+                except RuntimeError as err:
+                    raise UpgradeFailedError from err
+                else:
+                    if (
+                        hasattr(handler, "err_level")
+                        and handler.err_level >= gdal.CE_Warning
+                    ):
+                        warnings.append(handler.err_msg)
+
+            if len(warnings) > 0:
+                warning_string = "\n".join(warnings)
+                raise UpgradeFailedError(
+                    "GeoPackage conversion didn't finish as expected:\n", warning_string
+                )
+
         # Correct path of current database
         self.db.path = Path(self.db.path).with_suffix(".gpkg")
         # Reset engine so new path is used on the next call of get_engine()

threedi_schema/tests/test_gpkg.py

@@ -1,47 +1,32 @@
-import re
-import subprocess
-
 import pytest
 
+from sqlalchemy import text
+
 from threedi_schema.application.schema import get_schema_version
+from threedi_schema.infrastructure.spatialite_versions import get_spatialite_version
+
 
 
-def test_convert_to_geopackage(oldest_sqlite):
-    if get_schema_version() < 300:
-        pytest.skip("Gpkg not supported for schema < 300")
+@pytest.mark.parametrize("upgrade_spatialite", [True, False])
+def test_convert_to_geopackage(oldest_sqlite, upgrade_spatialite):
+    # if get_schema_version() < 300:
+    #     pytest.skip("Gpkg not supported for schema < 300")
     # In case the fixture changes and refers to a geopackage,
     # convert_to_geopackage will be ignored because the db is already a geopackage
-    assert oldest_sqlite.get_engine().dialect.name == "sqlite"
-    # check if ogr2ogr is installed and has the right version:
-    result = subprocess.run(
-        "ogr2ogr --version",
-        shell=True,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        text=True,
-    )
-    expect_success = False
-    if result.returncode == 0:
-        # get version
-        version = re.findall(r"\b(\d+\.\d+\.\d+)\b", result.stdout)[0]
-        # trim patch version and convert to float
-        float_version = float(version[0 : version.rfind(".")])
-        if float_version >= 3.4:
-            expect_success = True
-    if expect_success:
-        oldest_sqlite.schema.convert_to_geopackage()
-        # Ensure that after the conversion the geopackage is used
-        assert oldest_sqlite.path.suffix == ".gpkg"
-        assert oldest_sqlite.get_engine().dialect.name == "geopackage"
-        assert oldest_sqlite.schema.validate_schema()
-    else:
-        # Upgrade is not performed in convert_to_geopackage when ogr2ogr doesn't run
-        # because no operation should be performed in that case.
-        # However, this wil result in an invalid schema, so here we will run upgrade manually.
-        # Because convert_to_geopackage() is only used via upgrade; this is not an issue.
-        with pytest.warns():
-            oldest_sqlite.schema.convert_to_geopackage()
-            oldest_sqlite.schema.upgrade()
-            assert oldest_sqlite.path.suffix == ".sqlite"
-            assert oldest_sqlite.get_engine().dialect.name == "sqlite"
-            assert oldest_sqlite.schema.validate_schema()
+
+    if upgrade_spatialite:
+        _, file_version = get_spatialite_version(oldest_sqlite)
+        assert file_version == 3
+        oldest_sqlite.schema.upgrade(revision="0229")
+        oldest_sqlite.schema.upgrade_spatialite_version()
+        _, file_version = get_spatialite_version(oldest_sqlite)
+        assert file_version >= 4
+
+    oldest_sqlite.schema.convert_to_geopackage()
+    # Ensure that after the conversion the geopackage is used
+    assert oldest_sqlite.path.suffix == ".gpkg"
+    with oldest_sqlite.session_scope() as session:
+        gpkg_table_exists = bool(session.execute(text("SELECT count(*) FROM sqlite_master WHERE type='table' AND name='gpkg_contents';")).scalar())
+
+    assert gpkg_table_exists
+    assert oldest_sqlite.schema.validate_schema()