[Backport 5.0.x] [Fixes #14041] Allow upsert / replace for GPKG (#14073)

mattiagiupponi · web-flow · commit ffce5fd3bc38 · 2026-03-20T11:50:29.000+01:00
diff --git a/geonode/upload/handlers/common/tests_vector.py b/geonode/upload/handlers/common/tests_vector.py
@@ -519,34 +519,6 @@ def test_upsert_validation_should_fail(self):
             "The Dynamic model generation must be enabled to perform the upsert IMPORTER_ENABLE_DYN_MODELS=True",
         )
 
-    @override_settings(IMPORTER_ENABLE_DYN_MODELS=True)
-    @patch("geonode.upload.handlers.common.vector.ModelSchema")
-    @patch("geonode.upload.handlers.common.vector.BaseVectorFileHandler.extract_upsert_key")
-    def test_upsert_data_should_fail_if_upsertkey_is_not_provided(self, upsert_function, schema):
-        """
-        The test should fail since the upsert key provided is empty/Null and
-        was not possible to extract the key from the DB schema
-        """
-        schema.return_value = MagicMock()
-        data = create_single_dataset("example_upsert_dataset")
-        exec_id = orchestrator.create_execution_request(
-            user=self.user,
-            func_name="funct1",
-            step="step",
-            input_params={"files": self.valid_files, "skip_existing_layer": True, "resource_pk": data.pk},
-        )
-
-        upsert_function.return_value = None
-        handler = ShapeFileHandler()
-        with self.assertRaises(Exception) as exept:
-            handler.upsert_data(["files"], exec_id)
-
-        self.assertIsNotNone(exept)
-        self.assertEqual(
-            str(exept.exception),
-            "Was not possible to find the upsert key, upsert is aborted",
-        )
-
     def test_get_error_file_csv_headers(self):
         handler = BaseVectorFileHandler()
         mock_validator = MagicMock()
@@ -622,29 +594,6 @@ def test_upsert_data_without_dynamic_model_schema(self):
             "This dataset does't support updates. Please upload the dataset again to have the upsert operations enabled",
         )
 
-    def test_upsert_data_raise_error_if_upsert_key_is_not_defined(self):
-        """
-        Should raise error if the dynamic model schema is not present
-        """
-        data = create_single_dataset("example_upsert_dataset")
-        exec_id = orchestrator.create_execution_request(
-            user=self.user,
-            func_name="funct1",
-            step="step",
-            input_params={
-                "files": self.original,
-                "skip_existing_layer": True,
-                "resource_pk": data.pk,
-                "upsert_key": None,
-            },
-        )
-        ModelSchema.objects.create(name="example_upsert_dataset", db_name="datastore", managed=True)
-
-        with self.assertRaises(UpsertException) as exp:
-            self.handler.upsert_data(self.original, exec_id)
-
-        self.assertEqual(str(exp.exception), "Was not possible to find the upsert key, upsert is aborted")
-
     def test_validate_single_feature_raise_error(self):
         """
         Should raise error if the dynamic model schema is not present
diff --git a/geonode/upload/handlers/common/vector.py b/geonode/upload/handlers/common/vector.py
@@ -455,7 +455,7 @@ def import_resource(self, files: dict, execution_id: str, **kwargs) -> str:
         data inside the geonode_data database
         """
         all_layers = self.get_ogr2ogr_driver().Open(files.get("base_file"))
-        layers = self._select_valid_layers(all_layers)
+        layers = self._select_valid_layers(all_layers, execution_id=execution_id)
         # for the moment we skip the dyanamic model creation
         layer_count = len(layers)
         logger.info(f"Total number of layers available: {layer_count}")
@@ -556,7 +556,7 @@ def import_resource(self, files: dict, execution_id: str, **kwargs) -> str:
             raise e
         return layer_names, alternates, execution_id
 
-    def _select_valid_layers(self, all_layers):
+    def _select_valid_layers(self, all_layers, **kwargs):
         layers = []
         for layer in all_layers:
             try:
@@ -738,7 +738,17 @@ def _define_dynamic_layer_schema(self, layer, **kwargs):
                     "authority": self.identify_authority(layer),
                 }
             ]
-
+        # if the layer comes from a DB, the fid column is not included in the schema, but we need to add it as primary key for the dynamic model
+        fid_in_schema = any(x["name"] == DEFAULT_PK_COLUMN_NAME for x in layer_schema)
+        if not fid_in_schema and layer.GetFIDColumn():
+            layer_schema += [
+                {
+                    "name": layer.GetFIDColumn(),
+                    "class_name": "django.db.models.BigAutoField",
+                    "null": False,
+                    "primary_key": True,
+                }
+            ]
         return layer_schema
 
     def promote_to_multi(self, geometry_name):
@@ -1129,7 +1139,7 @@ def __get_new_and_original_schema(self, files, execution_id):
 
         # use ogr2ogr to read the uploaded files for the upsert
         all_layers = self.get_ogr2ogr_driver().Open(files.get("base_file"))
-        layers = self._select_valid_layers(all_layers)
+        layers = self._select_valid_layers(all_layers, execution_id=execution_id)
         if not layers:
             raise UpsertException("No valid layers found in the provided file for upsert.")
 
@@ -1213,20 +1223,20 @@ def upsert_data(self, files, execution_id, **kwargs):
         # get the rows that match the upsert key
         OriginalResource = model.as_model()
 
-        # retrieve the upsert key.
-        upsert_key = self.extract_upsert_key(exec_obj, dynamic_model_instance=model)
-        if not upsert_key:
-            # if for any reason the key is not present, better to raise an error
-            raise UpsertException("Was not possible to find the upsert key, upsert is aborted")
         # use ogr2ogr to read the uploaded files values for the upsert
         all_layers = self.get_ogr2ogr_driver().Open(files.get("base_file"))
         valid_create = 0
         valid_update = 0
-        layers = self._select_valid_layers(all_layers)
+        layers = self._select_valid_layers(all_layers, execution_id=execution_id)
         if not layers:
             raise UpsertException("No valid layers were found in the file provided")
         # we can upsert just 1 layer at time
 
+        upsert_key = self.extract_upsert_key(layers[0])
+        if not upsert_key:
+            # if for any reason the key is not present, better to raise an error
+            raise UpsertException("Was not possible to find the upsert key, upsert is aborted")
+
         self._validate_single_feature(exec_obj, OriginalResource, upsert_key, layers, iter(layers[0]))
 
         valid_create, valid_update = self._commit_upsert(model, OriginalResource, upsert_key, iter(layers[0]))
@@ -1359,9 +1369,16 @@ def _validate_feature(self, data_chunk, model_instance, upsert_key, errors):
     def _save_feature(self, data_chunk, model_obj, model_instance, upsert_key, valid_update, valid_create):
         # getting all the upsert_key value from the data chunk
         # retrieving the data from the DB
-        value_in_db = model_instance.objects.filter(
-            **{f"{upsert_key}__in": (getattr(feature, upsert_key) for feature in data_chunk)}
-        ).in_bulk(field_name=upsert_key)
+        use_get_fid = False  # flag to understand if we need to use the GetFID as upsert key, this is needed for DB drivers with FID columns that hide the FID field from the schema
+        filters = []
+        for feature in data_chunk:
+            # DB drivers with FID columns hide the FID field from the schema, so we need to check if the FID is present and use it as upsert key if the upsert key is the default one
+            if not getattr(feature, upsert_key, None) and feature.GetFID() != ogr.NullFID:
+                filters.append(feature.GetFID())
+                use_get_fid = True
+            else:
+                filters.append(getattr(feature, upsert_key))
+        value_in_db = model_instance.objects.filter(**{f"{upsert_key}__in": filters}).in_bulk(field_name=upsert_key)
         # looping over the chunk data
         to_process = []
         feature_to_save = []
@@ -1384,6 +1401,8 @@ def _save_feature(self, data_chunk, model_obj, model_instance, upsert_key, valid
             feature_as_dict.update(
                 {self.default_geometry_column_name: f"SRID={code};{self.promote_geom_to_multi(geom).ExportToWkt()}"}
             )
+            if use_get_fid:
+                feature_as_dict[upsert_key] = feature.GetFID()
             to_process.append(feature_as_dict)
 
         for feature_as_dict in to_process:
@@ -1419,18 +1438,13 @@ def validate_feature(self, feature):
             feature["error"] = " | ".join(errors)
             return feature, False
 
-    def extract_upsert_key(self, exec_obj, dynamic_model_instance):
-        # first we check if the upsert key is passed by the call
-        key = exec_obj.input_params.get("upsert_key", DEFAULT_PK_COLUMN_NAME)
-        if not key:
-            # if the upsert key is not passed, we use the primary key as upsert key
-            # the primary key is defined in the Fields of the dynamic model
-            # dynamic models raise error if we filter the json with ORM
-            key = [x.name for x in dynamic_model_instance.fields.all() if x.kwargs.get("primary_key")]
-            if key:
-                return key[0]
+    def extract_upsert_key(self, layer):
 
-        return key
+        fid_in_schema = any(x.name == DEFAULT_PK_COLUMN_NAME for x in layer.schema)
+        if not fid_in_schema and layer.GetFIDColumn():
+            return layer.GetFIDColumn()
+
+        return DEFAULT_PK_COLUMN_NAME
 
     def refresh_geonode_resource(self, execution_id, asset=None, dataset=None, create_asset=True, **kwargs):
         # getting execution_id information
@@ -1485,7 +1499,6 @@ def fixup_dynamic_model_fields(self, _exec, files, **kwargs):
             self.__fixup_primary_key(dataset)
 
 
-
 @importer_app.task(
     base=ErrorBaseTaskClass,
     name="geonode.upload.import_next_step",
diff --git a/geonode/upload/handlers/gpkg/handler.py b/geonode/upload/handlers/gpkg/handler.py
@@ -18,7 +18,6 @@
 #########################################################################
 import logging
 
-from geonode.resource.enumerator import ExecutionRequestAction as exa
 from geonode.upload.api.exceptions import UploadParallelismLimitException
 from geonode.upload.utils import UploadLimitValidator
 from geopackage_validator.validate import validate
@@ -27,6 +26,7 @@
 
 from geonode.upload.handlers.common.vector import BaseVectorFileHandler
 from geonode.upload.utils import ImporterRequestAction as ira
+from geonode.upload.orchestrator import orchestrator
 
 logger = logging.getLogger("importer")
 
@@ -37,26 +37,6 @@ class GPKGFileHandler(BaseVectorFileHandler):
     It must provide the task_lists required to comple the upload
     """
 
-    TASKS = {
-        exa.UPLOAD.value: (
-            "start_import",
-            "geonode.upload.import_resource",
-            "geonode.upload.publish_resource",
-            "geonode.upload.create_geonode_resource",
-        ),
-        exa.COPY.value: (
-            "start_copy",
-            "geonode.upload.copy_dynamic_model",
-            "geonode.upload.copy_geonode_data_table",
-            "geonode.upload.publish_resource",
-            "geonode.upload.copy_geonode_resource",
-        ),
-        ira.ROLLBACK.value: (
-            "start_rollback",
-            "geonode.upload.rollback",
-        ),
-    }
-
     @property
     def supported_file_extension_config(self):
         return {
@@ -161,3 +141,12 @@ def handle_xml_file(self, saved_dataset, _exec):
         Not implemented for GPKG, skipping
         """
         pass
+
+    def _select_valid_layers(self, all_layers, **kwargs):
+        layers = super()._select_valid_layers(all_layers=all_layers)
+        if execution_id := kwargs.get("execution_id", None):
+            exec_obj = orchestrator.get_execution_object(execution_id)
+            if exec_obj.action in (ira.REPLACE.value, ira.UPSERT.value):
+                if len(layers) > 1:
+                    raise InvalidGeopackageException("For Upsert and Replace, only one layer is allowed in the GPKG")
+        return layers
diff --git a/geonode/upload/handlers/gpkg/tests.py b/geonode/upload/handlers/gpkg/tests.py
@@ -17,7 +17,7 @@
 #
 #########################################################################
 import shutil
-
+import os
 from django.test import TestCase, override_settings
 from django.contrib.auth import get_user_model
 from geonode.upload.handlers.gpkg.handler import GPKGFileHandler
@@ -41,6 +41,7 @@ def setUpClass(cls):
         cls.handler = GPKGFileHandler()
         cls.valid_gpkg = f"{project_dir}/tests/fixture/valid.gpkg"
         cls.invalid_gpkg = f"{project_dir}/tests/fixture/invalid.gpkg"
+        cls.multiple_layer = f"{project_dir}/tests/fixture/multiple_layer.gpkg"
         cls.user, _ = get_user_model().objects.get_or_create(username="admin")
         cls.invalid_files = {"base_file": cls.invalid_gpkg}
         cls.valid_files = {"base_file": cls.valid_gpkg, "action": "upload"}
@@ -164,3 +165,36 @@ def test_single_message_error_handler(self):
         # Assert the error was recorded
         errors = exec_request_obj.output_params.get("errors", [])
         assert any("exception raised" in str(e) for e in errors)
+
+    def test_select_valid_layers(self):
+        """
+        The function should return only the datasets with a geometry
+        The other one are discarded
+        """
+        # Copy the file to the temporary folder
+        shutil.copy(self.multiple_layer, "/tmp")
+
+        exec_id = orchestrator.create_execution_request(
+            user=get_user_model().objects.first(),
+            func_name="funct1",
+            step="step",
+            action="replace",
+            input_params={
+                "files": {"base_file": "/tmp/multiple_layer.gpkg"},
+                "skip_existing_layer": True,
+                "store_spatial_file": False,
+                "handler_module_path": str(self.handler),
+            },
+        )
+
+        all_layers = GPKGFileHandler().get_ogr2ogr_driver().Open("/tmp/multiple_layer.gpkg")
+
+        with self.assertRaises(Exception) as exp:
+            GPKGFileHandler()._select_valid_layers(all_layers, execution_id=str(exec_id))
+
+        self.assertIn(
+            "For Upsert and Replace, only one layer is allowed in the GPKG",
+            exp.exception.args[0],
+        )
+
+        os.remove("/tmp/multiple_layer.gpkg")
diff --git a/geonode/upload/tests/fixture/multiple_layer.gpkg b/geonode/upload/tests/fixture/multiple_layer.gpkg
