Allow using spatialite function on geopackge in dbmanager

Enables Spatialite's amphibious mode for Geopackages, allowing the use of geospatial functions on geopackages in the DB manager.
qgis · Nov 15, 2019 · 73023e8 · 73023e8
1 parent f53e137
commit 73023e8
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 0 deletions.
diff --git a/python/utils.py b/python/utils.py
@@ -677,6 +677,13 @@ def fcnRegexp(pattern, string):
             break
     if not found:
         raise RuntimeError("Cannot find any suitable spatialite module")
+    if any(['.gpkg' in arg for arg in args]):
+        try:
+            cur.execute("SELECT EnableGpkgAmphibiousMode()")
+        except (sqlite3.Error, sqlite3.DatabaseError, sqlite3.NotSupportedError):
+            QgsMessageLog.logMessage(u"warning:{}".format("Could not enable geopackage amphibious mode"),
+                                     QCoreApplication.translate("Python", "Python warning"))
+
     cur.close()
     con.enable_load_extension(False)
     con.create_function("regexp", 2, fcnRegexp)

diff --git a/tests/src/python/test_db_manager_gpkg.py b/tests/src/python/test_db_manager_gpkg.py
@@ -24,6 +24,8 @@
 from plugins.db_manager.db_plugins import supportedDbTypes, createDbPlugin
 from plugins.db_manager.db_plugins.plugin import TableField
 
+from utilities import unitTestDataPath
+
 
 def GDAL_COMPUTE_VERSION(maj, min, rev):
     return ((maj) * 1000000 + (min) * 10000 + (rev) * 100)
@@ -444,6 +446,22 @@ def testAllGeometryTypes(self):
 
         connection.remove()
 
+    def testAmphibiousMode(self,):
+        connectionName = 'geopack1'
+        plugin = createDbPlugin('gpkg')
+        uri = QgsDataSourceUri()
+        test_gpkg = os.path.join(os.path.join(unitTestDataPath(), 'provider'), 'test_json.gpkg')
+
+        uri.setDatabase(test_gpkg)
+        plugin.addConnection(connectionName, uri)
+        connection = createDbPlugin('gpkg', connectionName)
+        connection.connect()
+        db = connection.database()
+        res = db.connector._execute(None, "SELECT St_area({}) from foo".format(db.tables()[0].fields()[1].name))
+        results = [row for row in res]
+        self.assertEqual(results, [(215229.265625,), (247328.171875,), (261752.78125,), (547597.2109375,), (15775.7578125,), (101429.9765625,), (268597.625,), (1634833.390625,), (596610.3359375,), (5268.8125,)])
+        connection.remove()
+
 
 if __name__ == '__main__':
     unittest.main()