From 1762b4a7b1bb1cde7fcf13321bb2cbad29280318 Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Mon, 20 Jul 2026 11:26:47 +0000 Subject: [PATCH 1/4] added graceful error handling for tehsils without any village with village id --- utilities/pipelines/tabular.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/utilities/pipelines/tabular.py b/utilities/pipelines/tabular.py index 29fb3976..f3aad2c4 100644 --- a/utilities/pipelines/tabular.py +++ b/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) From db96f08d69811a6b5f7ab7c0dea4300543bf9543 Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Mon, 20 Jul 2026 11:28:16 +0000 Subject: [PATCH 2/4] resolved add to db, corrected db schema to match with the schema accepted by save_layer_info_to_db, update_layer_sync_status functions --- utilities/pipelines/publish.py | 82 ++++++++++++++++++++++------------ 1 file changed, 53 insertions(+), 29 deletions(-) diff --git a/utilities/pipelines/publish.py b/utilities/pipelines/publish.py index ea8f6a6b..03dcaa9e 100644 --- a/utilities/pipelines/publish.py +++ b/utilities/pipelines/publish.py @@ -421,13 +421,9 @@ def register_layer( ) -> 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; + 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() @@ -437,32 +433,60 @@ def register_layer( 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 + from django.db import transaction + + from computing.models import Dataset, Layer, LayerType + from geoadmin.models import DistrictSOI, StateSOI, TehsilSOI - Dataset.objects.get_or_create( + with transaction.atomic(): + dataset, _ = Dataset.objects.get_or_create( name=dataset_name, - defaults={"layer_type": LayerType.VECTOR, "workspace": workspace}, + 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, ) - layer_id = save_layer_info_to_db( - state=state, - district=district, - block=block, + block_obj = TehsilSOI.objects.get( + tehsil_name__iexact=block, + district=district_obj, + ) + layer, _ = Layer.objects.update_or_create( + dataset=dataset, 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, + 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 {}), + }, + }, ) - 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]} + return { + "ok": True, + "status": "registered", + "dataset": dataset_name, + "layer_id": layer.id, + } From 593372967074075942e6b086d01def6939469a3a Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Mon, 20 Jul 2026 11:29:37 +0000 Subject: [PATCH 3/4] revised save to db usage --- computing/misc/livestocks/pipeline.py | 46 +++++++++++++-------------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/computing/misc/livestocks/pipeline.py b/computing/misc/livestocks/pipeline.py index d8de86a7..97ad3459 100644 --- a/computing/misc/livestocks/pipeline.py +++ b/computing/misc/livestocks/pipeline.py @@ -486,29 +486,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 +497,29 @@ 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"), + 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"]["layer_id"] if outputs.readme: result["readme_path"] = bundle.write_readme( _readme_lines( From 7aa32d94de5b4750d602afee6e16a384505f38ff Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Mon, 20 Jul 2026 11:30:10 +0000 Subject: [PATCH 4/4] Use canonical DB helpers for local pipeline registration --- computing/misc/antyodaya/pipeline.py | 62 +++++++++++-------- computing/misc/facilities/pipeline.py | 59 ++++++++++++------- computing/misc/livestocks/pipeline.py | 52 ++++++++++------ utilities/pipelines/__init__.py | 2 - utilities/pipelines/publish.py | 85 --------------------------- 5 files changed, 108 insertions(+), 152 deletions(-) diff --git a/computing/misc/antyodaya/pipeline.py b/computing/misc/antyodaya/pipeline.py index baeace53..3e5aa107 100644 --- a/computing/misc/antyodaya/pipeline.py +++ b/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( diff --git a/computing/misc/facilities/pipeline.py b/computing/misc/facilities/pipeline.py index 413b3fec..7e847551 100644 --- a/computing/misc/facilities/pipeline.py +++ b/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") diff --git a/computing/misc/livestocks/pipeline.py b/computing/misc/livestocks/pipeline.py index 97ad3459..bd2b2d71 100644 --- a/computing/misc/livestocks/pipeline.py +++ b/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( diff --git a/utilities/pipelines/__init__.py b/utilities/pipelines/__init__.py index a0960d1e..1bc80a6a 100644 --- a/utilities/pipelines/__init__.py +++ b/utilities/pipelines/__init__.py @@ -7,7 +7,6 @@ """ from .admin import AdminScope, CSAdminSource -from .publish import register_layer from .schema import StandardRequest, api_request_payload, load_config __all__ = [ @@ -15,6 +14,5 @@ "CSAdminSource", "StandardRequest", "api_request_payload", - "register_layer", "load_config", ] diff --git a/utilities/pipelines/publish.py b/utilities/pipelines/publish.py index 03dcaa9e..f697a5af 100644 --- a/utilities/pipelines/publish.py +++ b/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, - }