Fix local pipeline database registration and empty village-ID handling

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,
@@ -486,29 +487,6 @@ def run_livestocks_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", "Livestock Census"),
-                        layer_name=layer_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": 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,
-                    )
-                    result["layer_id"] = (result["layer_registration"] or {}).get("layer_id")
             except Exception as exc:
                 geoserver = {
                     "ok": False,
@@ -520,6 +498,42 @@ 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"):
+        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,
+            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 {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,64 +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.
-
-    Uses the same `save_layer_info_to_db` path as the GEE-backed pipelines, so
-    local pipeline layers appear alongside every other layer. 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.
-
-    Never raises: the run has already succeeded by the time this is called, so
-    a database problem is reported in the result instead of losing the outputs.
-    """
-
-    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"}
-
-    try:
-        from computing.models import Dataset, LayerType
-        from computing.utils import save_layer_info_to_db, update_layer_sync_status
-
-        Dataset.objects.get_or_create(
-            name=dataset_name,
-            defaults={"layer_type": LayerType.VECTOR, "workspace": workspace},
-        )
-        layer_id = save_layer_info_to_db(
-            state=state,
-            district=district,
-            block=block,
-            layer_name=layer_name,
-            asset_id=geoserver_url,
-            dataset_name=dataset_name,
-            algorithm=algorithm,
-            algorithm_version=algorithm_version,
-            misc={"is_generated_locally": True, "geoserver_workspace": workspace, **(misc or {})},
-            is_override=overwrite,
-            is_gee_asset=False,
-        )
-        if not layer_id:
-            return {"ok": False, "status": "not_registered", "dataset": dataset_name,
-                    "reason": "save_layer_info_to_db returned no layer id (check state/district/block exist in the SOI tables)"}
-        update_layer_sync_status(layer_id=layer_id, sync_to_geoserver=True)
-        return {"ok": True, "status": "registered", "dataset": dataset_name, "layer_id": layer_id}
-    except Exception as exc:
-        return {"ok": False, "status": "registration_failed", "dataset": dataset_name,
-                "error_type": exc.__class__.__name__, "error": str(exc)[:500]}

utilities/pipelines/tabular.py

@@ -135,7 +135,12 @@ def fetch_by_values(
 
         values = [value for value in values if value is not None and value == value]
         if not values:
-            return pd.DataFrame(columns=list(columns or []))
+            empty_columns = (
+                columns
+                if columns is not None
+                else (*self.key_columns, *(self.source_columns or ()))
+            )
+            return pd.DataFrame(columns=list(dict.fromkeys(empty_columns)))
         if not self.is_fresh():
             self.materialize()
         select_columns = "*" if columns is None else ", ".join(quote_identifier(col) for col in columns)