Use canonical DB helpers for local pipeline registration

Author: amit-spatial

computing/misc/antyodaya/pipeline.py

@@ -13,6 +13,7 @@
 import pandas as pd
 from django.conf import settings
 
+from computing.utils import save_layer_info_to_db, update_layer_sync_status
 from utilities.pipelines import AdminScope, CSAdminSource, StandardRequest, load_config
 from utilities.pipelines.admin import (
     ADMIN_COLUMN_DESCRIPTIONS,
@@ -28,7 +29,7 @@
     stable_hash,
     utc_now_text,
 )
-from utilities.pipelines.publish import publish_gpkg_layer, register_layer
+from utilities.pipelines.publish import publish_gpkg_layer
 from utilities.pipelines.schema import (
     STATUS_MATCHED,
     STATUS_NO_DATA,
@@ -597,29 +598,6 @@ def run_antyodaya_pipeline(
                 geoserver = asdict(geoserver_result)
                 geoserver["ok"] = True
                 geoserver["status"] = "published"
-                if request.publish.register_layers:
-                    result["layer_registration"] = register_layer(
-                        dataset_name=output_config.get("dataset_name", "Antyodaya 2020"),
-                        layer_name=result_name,
-                        scope=request.scope,
-                        workspace=geoserver_workspace,
-                        geoserver_url=geoserver.get("wfs_url"),
-                        algorithm=ALGORITHM,
-                        algorithm_version=ALGORITHM_VERSION,
-                        misc={
-                            "source_csv": config["sources"]["csv"],
-                            "gpkg_path": result.get("gpkg_path"),
-                            "links_path": result.get("links_path"),
-                            "output_dir": bundle.path.as_posix(),
-                            "geoserver_layer_name": result_name,
-                            "geoserver_url": geoserver.get("wfs_url"),
-                            "rows": result.get("rows"),
-                            "matched_rows": result.get("matched_rows"),
-                            "join_coverage": result.get("join_coverage"),
-                        },
-                        overwrite=request.publish.overwrite,
-                    )
-                    result["layer_id"] = (result["layer_registration"] or {}).get("layer_id")
             except Exception as exc:
                 geoserver = {
                     "ok": False,
@@ -631,6 +609,42 @@ def run_antyodaya_pipeline(
                 }
         timings["publish_geoserver_seconds"] = round(time.perf_counter() - t0, 3)
     result["geoserver"] = geoserver
+    if request.publish.register_layers and geoserver and geoserver.get("ok"):
+        state = request.scope.state_name
+        district = request.scope.district_name
+        block = request.scope.tehsil_name
+        if not (state and district and block):
+            raise ValueError(
+                "Layer registration requires state, district, and tehsil names."
+            )
+        dataset_name = output_config.get("dataset_name", "Antyodaya 2020")
+        layer_id = save_layer_info_to_db(
+            state=state,
+            district=district,
+            block=block,
+            layer_name=result_name,
+            asset_id="not applicable: local compute GeoServer layer",
+            dataset_name=dataset_name,
+            algorithm=ALGORITHM,
+            algorithm_version=ALGORITHM_VERSION,
+            misc={"is_generated_locally": True},
+            is_override=request.publish.overwrite,
+        )
+        if layer_id is None:
+            raise RuntimeError(f"Database registration failed for layer {result_name!r}.")
+        if update_layer_sync_status(
+            layer_id=layer_id, sync_to_geoserver=True
+        ) is None:
+            raise RuntimeError(
+                f"GeoServer sync status update failed for layer ID {layer_id}."
+            )
+        result["layer_id"] = layer_id
+        result["layer_registration"] = {
+            "ok": True,
+            "status": "registered",
+            "dataset": dataset_name,
+            "layer_id": layer_id,
+        }
     if outputs.readme:
         result["readme_path"] = bundle.write_readme(
             _readme_lines(

computing/misc/facilities/pipeline.py

@@ -16,6 +16,7 @@
 from django.conf import settings
 from scipy.spatial import cKDTree
 
+from computing.utils import save_layer_info_to_db, update_layer_sync_status
 from utilities.pipelines import AdminScope, CSAdminSource, StandardRequest, load_config
 from utilities.pipelines.admin import (
     ADMIN_COLUMN_DESCRIPTIONS,
@@ -53,7 +54,6 @@
 from utilities.pipelines.publish import (
     publish_gpkg_layer,
     publish_gpkg_layers,
-    register_layer,
 )
 from utilities.pipelines.unicode import normalize_unicode_frame
 from nrm_app.celery import app
@@ -1134,33 +1134,48 @@ def run_facilities_pipeline(
         }
     ).as_posix()
     if request.publish.register_layers and published_layers:
-        common_misc = {
-            "source_facilities_gpkg": config["sources"]["facilities_gpkg"],
-            "gpkg_path": result.get("gpkg_path"),
-            "facility_points_gpkg_path": result.get("facility_points_gpkg_path"),
-            "links_path": result.get("links_path"),
-            "output_dir": bundle.path.as_posix(),
-            "village_rows": result.get("village_rows"),
-            "nearest_rows": result.get("nearest_rows"),
-            "inventory_rows": result.get("inventory_rows"),
-        }
+        state = request.scope.state_name
+        district = request.scope.district_name
+        block = request.scope.tehsil_name
+        if not (state and district and block):
+            raise ValueError(
+                "Layer registration requires state, district, and tehsil names."
+            )
         registrations: dict[str, dict[str, Any]] = {}
         for role, published in published_layers.items():
-            registrations[role] = register_layer(
-                dataset_name=(
-                    output_config.get("dataset_name", "Facilities Proximity")
-                    if role == "village_properties"
-                    else output_config.get("points_dataset_name", "Facilities Points")
-                ),
+            dataset_name = (
+                output_config.get("dataset_name", "Facilities Proximity")
+                if role == "village_properties"
+                else output_config.get("points_dataset_name", "Facilities Points")
+            )
+            layer_id = save_layer_info_to_db(
+                state=state,
+                district=district,
+                block=block,
                 layer_name=published["layer_name"],
-                scope=request.scope,
-                workspace=published["workspace"],
-                geoserver_url=published["wfs_url"],
+                asset_id="not applicable: local compute GeoServer layer",
+                dataset_name=dataset_name,
                 algorithm=ALGORITHM,
                 algorithm_version=ALGORITHM_VERSION,
-                misc={**common_misc, "output_role": role},
-                overwrite=request.publish.overwrite,
+                misc={"is_generated_locally": True},
+                is_override=request.publish.overwrite,
             )
+            if layer_id is None:
+                raise RuntimeError(
+                    f"Database registration failed for layer {published['layer_name']!r}."
+                )
+            if update_layer_sync_status(
+                layer_id=layer_id, sync_to_geoserver=True
+            ) is None:
+                raise RuntimeError(
+                    f"GeoServer sync status update failed for layer ID {layer_id}."
+                )
+            registrations[role] = {
+                "ok": True,
+                "status": "registered",
+                "dataset": dataset_name,
+                "layer_id": layer_id,
+            }
         result["layer_registrations"] = registrations
         result["layer_registration"] = registrations.get("village_properties")
         result["layer_id"] = (result.get("layer_registration") or {}).get("layer_id")

computing/misc/livestocks/pipeline.py

@@ -12,6 +12,7 @@
 import pandas as pd
 from django.conf import settings
 
+from computing.utils import save_layer_info_to_db, update_layer_sync_status
 from utilities.pipelines import AdminScope, CSAdminSource, StandardRequest, load_config
 from utilities.pipelines.admin import (
     ADMIN_COLUMN_DESCRIPTIONS,
@@ -27,7 +28,7 @@
     stable_hash,
     utc_now_text,
 )
-from utilities.pipelines.publish import publish_gpkg_layer, register_layer
+from utilities.pipelines.publish import publish_gpkg_layer
 from utilities.pipelines.schema import (
     STATUS_MATCHED,
     STATUS_NO_DATA,
@@ -498,28 +499,41 @@ def run_livestocks_pipeline(
         timings["publish_geoserver_seconds"] = round(time.perf_counter() - t0, 3)
     result["geoserver"] = geoserver
     if request.publish.register_layers and geoserver and geoserver.get("ok"):
-        result["layer_registration"] = register_layer(
-            dataset_name=output_config.get("dataset_name", "Livestock Census"),
+        state = request.scope.state_name
+        district = request.scope.district_name
+        block = request.scope.tehsil_name
+        if not (state and district and block):
+            raise ValueError(
+                "Layer registration requires state, district, and tehsil names."
+            )
+        dataset_name = output_config.get("dataset_name", "Livestock Census")
+        layer_id = save_layer_info_to_db(
+            state=state,
+            district=district,
+            block=block,
             layer_name=layer_name,
-            scope=request.scope,
-            workspace=geoserver_workspace,
-            geoserver_url=geoserver.get("wfs_url"),
+            asset_id="not applicable: local compute GeoServer layer",
+            dataset_name=dataset_name,
             algorithm=ALGORITHM,
             algorithm_version=ALGORITHM_VERSION,
-            misc={
-                "source_csv": config["sources"]["csv"],
-                "gpkg_path": result.get("gpkg_path"),
-                "links_path": result.get("links_path"),
-                "output_dir": bundle.path.as_posix(),
-                "geoserver_layer_name": layer_name,
-                "geoserver_url": geoserver.get("wfs_url"),
-                "rows": result.get("rows"),
-                "matched_rows": result.get("matched_rows"),
-                "join_coverage": result.get("join_coverage"),
-            },
-            overwrite=request.publish.overwrite,
+            misc={"is_generated_locally": True},
+            is_override=request.publish.overwrite,
         )
-        result["layer_id"] = result["layer_registration"]["layer_id"]
+        if layer_id is None:
+            raise RuntimeError(f"Database registration failed for layer {layer_name!r}.")
+        if update_layer_sync_status(
+            layer_id=layer_id, sync_to_geoserver=True
+        ) is None:
+            raise RuntimeError(
+                f"GeoServer sync status update failed for layer ID {layer_id}."
+            )
+        result["layer_id"] = layer_id
+        result["layer_registration"] = {
+            "ok": True,
+            "status": "registered",
+            "dataset": dataset_name,
+            "layer_id": layer_id,
+        }
     if outputs.readme:
         result["readme_path"] = bundle.write_readme(
             _readme_lines(

utilities/pipelines/__init__.py

@@ -7,14 +7,12 @@
 """
 
 from .admin import AdminScope, CSAdminSource
-from .publish import register_layer
 from .schema import StandardRequest, api_request_payload, load_config
 
 __all__ = [
     "AdminScope",
     "CSAdminSource",
     "StandardRequest",
     "api_request_payload",
-    "register_layer",
     "load_config",
 ]

utilities/pipelines/publish.py

@@ -405,88 +405,3 @@ def publish_gpkg_layers(
             property_count=len(verification["properties"]),
         )
     return results
-
-
-def register_layer(
-    *,
-    dataset_name: str,
-    layer_name: str,
-    scope: Any,
-    workspace: str,
-    geoserver_url: str,
-    algorithm: str,
-    algorithm_version: str,
-    misc: dict[str, Any] | None = None,
-    overwrite: bool = False,
-) -> dict[str, Any]:
-    """Register a published layer in the Core Stack layer database.
-
-    The `Layer` model is keyed on state/district/block, so only tehsil-scoped runs can register;
-    other scopes are reported as skipped rather than failing the run.
-
-    """
-
-    level = str(getattr(scope, "level", "") or "").lower()
-    if level not in {"tehsil", "block"}:
-        return {"ok": False, "status": "skipped", "reason": f"layer registration needs a tehsil scope, got {level!r}"}
-    state, district, block = scope.state_name, scope.district_name, scope.tehsil_name
-    if not (state and district and block):
-        return {"ok": False, "status": "skipped", "reason": "scope is missing state, district, or tehsil name"}
-
-    from django.db import transaction
-
-    from computing.models import Dataset, Layer, LayerType
-    from geoadmin.models import DistrictSOI, StateSOI, TehsilSOI
-
-    with transaction.atomic():
-        dataset, _ = Dataset.objects.get_or_create(
-            name=dataset_name,
-            defaults={
-                "layer_type": LayerType.VECTOR,
-                "workspace": workspace,
-                "is_active": True,
-            },
-        )
-        dataset.layer_type = LayerType.VECTOR
-        dataset.workspace = workspace
-        dataset.is_active = True
-        dataset.save(update_fields=["layer_type", "workspace", "is_active", "updated_at"])
-
-        state_obj = StateSOI.objects.get(state_name__iexact=state)
-        district_obj = DistrictSOI.objects.get(
-            district_name__iexact=district,
-            state=state_obj,
-        )
-        block_obj = TehsilSOI.objects.get(
-            tehsil_name__iexact=block,
-            district=district_obj,
-        )
-        layer, _ = Layer.objects.update_or_create(
-            dataset=dataset,
-            layer_name=layer_name,
-            state=state_obj,
-            district=district_obj,
-            block=block_obj,
-            layer_version="1.0",
-            defaults={
-                "algorithm": algorithm,
-                "algorithm_version": algorithm_version,
-                "is_sync_to_geoserver": True,
-                "is_override": overwrite,
-                "gee_asset_path": "not applicable: local compute GeoServer layer",
-                "is_public_gee_asset": False,
-                "misc": {
-                    "is_generated_locally": True,
-                    "source_type": "local_compute",
-                    "geoserver_workspace": workspace,
-                    "geoserver_url": geoserver_url,
-                    **(misc or {}),
-                },
-            },
-        )
-    return {
-        "ok": True,
-        "status": "registered",
-        "dataset": dataset_name,
-        "layer_id": layer.id,
-    }