[Fixes #13609] Tracking the failures of OGR2OGR related tasks (#13613)

* adding 'Multi 3D Polygon' geometry type

* tracking the failures of the ogr2ogr tasks

* removing the alternate as an extra key

* fixing flake issues

* excluding the 3d in the promote_to_multi

* fixing formatting issues

Author: Gpetrak

geonode/upload/celery_tasks.py

@@ -192,6 +192,39 @@ def set_task_status(self, task_name, execution_id, layer_key, status):
         )
 
 
+class UpdateDynamicTaskClass(Task):
+    max_retries = 3
+    track_started = True
+
+    def on_failure(self, exc, task_id, args, kwargs, einfo):
+        """
+        Called when the task fails.
+        Updates the ExecutionRequest.tasks dict and delegates logging/error handling to evaluate_error.
+        """
+        task_name = self.name
+        execution_id = args[0]
+        layer_key = find_key_recursively(kwargs, "layer_key")
+
+        _exec = orchestrator.get_execution_object(execution_id)
+        tasks_status = _exec.tasks or {}
+
+        if layer_key is not None:
+            # Ensure the layer exists
+            if layer_key not in tasks_status:
+                tasks_status[layer_key] = {}
+
+            tasks_status[layer_key][task_name] = "FAILED"
+
+        # Update ExecutionRequest tasks dict first
+        orchestrator.update_execution_request_status(
+            execution_id=execution_id,
+            tasks=tasks_status,
+        )
+
+        # Delegate the rest (errors, failed_layers, status) to evaluate_error
+        evaluate_error(self, exc, task_id, args, kwargs, einfo)
+
+
 @importer_app.task(
     bind=True,
     base=ErrorBaseTaskClass,
@@ -646,7 +679,7 @@ def copy_geonode_resource(self, exec_id, actual_step, layer_name, alternate, han
 
 
 @importer_app.task(
-    base=ErrorBaseTaskClass,
+    base=UpdateDynamicTaskClass,
     name="geonode.upload.create_dynamic_structure",
     queue="geonode.upload.create_dynamic_structure",
     max_retries=1,
@@ -660,6 +693,7 @@ def create_dynamic_structure(
     dynamic_model_schema_id: int,
     overwrite: bool,
     layer_name: str,
+    **kwargs,
 ):
     def _create_field(dynamic_model_schema, field, _kwargs):
         # common method to define the Field Schema object

geonode/upload/handlers/common/vector.py

@@ -43,12 +43,18 @@
 from geonode.base.models import ResourceBase
 from geonode.resource.enumerator import ExecutionRequestAction as exa
 from geonode.layers.models import Dataset
-from geonode.upload.celery_tasks import ErrorBaseTaskClass, FieldSchema, create_dynamic_structure
+from geonode.upload.celery_tasks import (
+    ErrorBaseTaskClass,
+    FieldSchema,
+    create_dynamic_structure,
+    UpdateDynamicTaskClass,
+)
 from geonode.upload.handlers.base import BaseHandler
 from geonode.upload.handlers.utils import (
     GEOM_TYPE_MAPPING,
     STANDARD_TYPE_MAPPING,
     drop_dynamic_model_schema,
+    create_layer_key,
 )
 from geonode.resource.manager import resource_manager
 from geonode.resource.models import ExecutionRequest
@@ -652,7 +658,14 @@ def create_dynamic_model_fields(
         # definition of the celery group needed to run the async workflow.
         # in this way each task of the group will handle only 30 field
         celery_group = group(
-            create_dynamic_structure.s(execution_id, schema, dynamic_model_schema.id, overwrite, layer_name)
+            create_dynamic_structure.s(
+                execution_id,
+                schema,
+                dynamic_model_schema.id,
+                overwrite,
+                layer_name,
+                layer_key=create_layer_key(layer.GetName(), str(execution_id)),
+            )
             for schema in list_chunked
         )
 
@@ -921,6 +934,7 @@ def get_ogr2ogr_task_group(
             handler_module_path,
             should_be_overwritten,
             alternate,
+            layer_key=create_layer_key(layer.lower(), str(execution_id)),
         )
 
     def _get_execution_request_object(self, execution_id: str):
@@ -1391,7 +1405,7 @@ def import_next_step(
 
 
 @importer_app.task(
-    base=ErrorBaseTaskClass,
+    base=UpdateDynamicTaskClass,
     name="geonode.upload.import_with_ogr2ogr",
     queue="geonode.upload.import_with_ogr2ogr",
     max_retries=1,
@@ -1406,6 +1420,7 @@ def import_with_ogr2ogr(
     handler_module_path: str,
     ovverwrite_layer=False,
     alternate=None,
+    **kwargs,
 ):
     """
     Perform the ogr2ogr command to import he gpkg inside geonode_data

geonode/upload/handlers/shapefile/handler.py

@@ -212,7 +212,7 @@ def promote_to_multi(self, geometry_name):
         Needed for the shapefiles
         Later this is used to map the geometry coming from ogr2ogr with a django class
         """
-        if "Multi" not in geometry_name and "Point" not in geometry_name:
+        if "Multi" not in geometry_name and "Point" not in geometry_name and "3D" not in geometry_name:
             return f"Multi {geometry_name.title()}"
         return geometry_name