from __future__ import annotations import json import pathlib import re from dataclasses import dataclass from hashlib import sha256 from datetime import datetime, timezone from math import isfinite from pathlib import Path from typing import Any from uuid import UUID import uuid from fastapi import UploadFile from shapely.geometry import MultiPoint, shape from sqlalchemy.orm import Session from app.core.errors import AppError from app.models import Area, Dataset, DatasetVersion, Project from app.services.data_contract_validation import ( ContractKind, DataAssetValidationInput, GeometryRecord, LineageEvidence, LineageStatus, ProvenanceStatus, QuarantineStatus, RASTER_GEOTIFF_CONTRACT_KEY, RASTER_GEOTIFF_CONTRACT_VERSION, TransformationEvidence, VECTOR_GEOJSON_CONTRACT_KEY, VECTOR_GEOJSON_CONTRACT_VERSION, ValidationIssue, ValidationReport, ValidationStatus, build_raster_ingest_input, build_vector_ingest_input, validate_registered_asset, ) from app.services.data_quarantine_service import DataQuarantineService from app.schemas.dataset import ( DatasetCreateResponse, DatasetStorageResponse, DatasetTemporalUpdate, DatasetVectorSummary, DatasetVersionRead, ) from app.services.geojson_service import parse_geojson_payload, load_dataset_text from app.services.raster_service import extract_raster_metadata from app.services.source_registry_service import SourceRegistryService from app.services.storage_service import StorageService from app.services.vector_feature_service import VectorFeatureService _CHECKSUM_SHA256 = re.compile(r"^[0-9a-f]{64}$") @dataclass(frozen=True) class _SourceVectorSchema: """Server-owned vector schema expectations attached to a source registry row.""" source_key: str expected_geometry_types: frozenset[str] required_attributes: tuple[str, ...] @classmethod def from_source(cls, source: Any) -> "_SourceVectorSchema": geometry_values = getattr(source, "expected_geometry_types_json", ()) expected_geometry_types = ( frozenset(str(value).strip() for value in geometry_values if str(value).strip()) if isinstance(geometry_values, (list, tuple, set)) else frozenset() ) attributes = getattr(source, "expected_attributes_json", {}) required_values = attributes.get("required") if isinstance(attributes, dict) else () if isinstance(required_values, str): required_values = (required_values,) required_attributes = ( tuple(sorted({str(value).strip() for value in required_values if str(value).strip()})) if isinstance(required_values, (list, tuple, set)) else () ) return cls( source_key=str(getattr(source, "source_key", "") or "").strip().lower(), expected_geometry_types=expected_geometry_types, required_attributes=required_attributes, ) def to_metadata(self, *, checked_feature_count: int) -> dict[str, Any]: return { "status": "passed", "source_key": self.source_key, "expected_geometry_types": sorted(self.expected_geometry_types), "required_attributes": list(self.required_attributes), "checked_feature_count": checked_feature_count, } def _feature_source_identifier(feature: dict[str, Any], properties: dict[str, Any]) -> Any: """Return the source identity under the GeoJSON and registry conventions.""" return feature.get("id") or properties.get("id") or properties.get("source_feature_id") def _validate_vector_feature_source_schema( *, feature: dict[str, Any], properties: dict[str, Any], geometry: Any, schema: _SourceVectorSchema, feature_context: str, ) -> None: """Fail closed when a source-specific vector expectation is violated.""" if schema.expected_geometry_types and geometry.geom_type not in schema.expected_geometry_types: raise AppError( code="SOURCE_SCHEMA_GEOMETRY_TYPE_NOT_ALLOWED", message=( f"{feature_context} has geometry type {geometry.geom_type}, which is not " f"allowed by source registry {schema.source_key or 'unknown'}" ), details={ "source_key": schema.source_key, "expected_geometry_types": sorted(schema.expected_geometry_types), "observed_geometry_type": geometry.geom_type, }, status_code=400, ) for attribute in schema.required_attributes: value = _feature_source_identifier(feature, properties) if attribute == "id" else properties.get(attribute) if value is None or (isinstance(value, str) and not value.strip()): raise AppError( code="SOURCE_SCHEMA_REQUIRED_ATTRIBUTE_MISSING", message=( f"{feature_context} is missing required source attribute {attribute!r} " f"for registry {schema.source_key or 'unknown'}" ), details={"source_key": schema.source_key, "required_attribute": attribute}, status_code=400, ) @dataclass(frozen=True) class _PartitionedVectorAudit: """Aggregate evidence from a full per-feature partition audit. The importer materializes one GeoJSON partition at a time because the current parser is ``json.loads`` based. It never materializes every regional partition or every regional Shapely geometry at once. """ feature_count: int geometry_types: tuple[str, ...] bounds_json: dict[str, float] partition_checksums_sha256: dict[str, str] source_schema_validation: dict[str, Any] representative_record: GeometryRecord def to_metadata(self) -> dict[str, Any]: return { "feature_count": self.feature_count, "geometry_types": list(self.geometry_types), "bounds_json": dict(self.bounds_json), "partition_checksums_sha256": dict(self.partition_checksums_sha256), "source_schema_validation": dict(self.source_schema_validation), "validation_mode": "partition_bounded_per_feature_with_aggregate_contract_record", } class _PartitionedGeoJsonRecords: """Perform a full, partition-bounded feature audit across GeoJSON partitions. The generic vector contract materializes its supplied geometry records. A regional artifact can contain hundreds of thousands of features, so this class validates one materialized partition at a time and emits a compact aggregate record for the generic source/checksum/CRS/bounds contract. Memory is bounded to the largest single partition, not to one feature. """ def __init__( self, partition_paths: list[str | Path], *, expected_feature_count: int, declared_partition_checksums: dict[str, Any] | None, source_schema: _SourceVectorSchema | None = None, ) -> None: self._partition_paths = tuple(Path(path) for path in partition_paths) self._expected_feature_count = expected_feature_count self._declared_checksums = declared_partition_checksums self._source_schema = source_schema or _SourceVectorSchema( source_key="", expected_geometry_types=frozenset(), required_attributes=(), ) def audit(self) -> _PartitionedVectorAudit: declared_checksums = self._validated_declared_checksums() observed_checksums: dict[str, str] = {} feature_count = 0 source_feature_ids: set[str] = set() geometry_types: set[str] = set() min_x: float | None = None min_y: float | None = None max_x: float | None = None max_y: float | None = None for partition_path in self._partition_paths: try: raw = partition_path.read_bytes() payload = json.loads(raw.decode("utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise AppError( code="INVALID_GEOJSON_PARTITION", message=f"Could not read GeoJSON partition {partition_path.name}", status_code=400, ) from exc features = payload.get("features") if isinstance(payload, dict) else None if not isinstance(payload, dict) or payload.get("type") != "FeatureCollection" or not isinstance(features, list): raise AppError( code="INVALID_GEOJSON_PARTITION", message=f"GeoJSON partition {partition_path.name} must be a FeatureCollection", status_code=400, ) observed_checksum = sha256(raw).hexdigest() expected_checksum = declared_checksums[partition_path.name] if observed_checksum != expected_checksum: raise AppError( code="PARTITION_CHECKSUM_MISMATCH", message=( f"Checksum for partition {partition_path.name} does not match " "the governed acquisition manifest." ), status_code=400, ) observed_checksums[partition_path.name] = observed_checksum for index, feature in enumerate(features): if not isinstance(feature, dict): raise AppError( code="INVALID_GEOJSON_PARTITION", message=f"Feature {index} in {partition_path.name} must be an object", status_code=400, ) properties = feature.get("properties") if isinstance(feature.get("properties"), dict) else {} source_feature_id = _feature_source_identifier(feature, properties) if source_feature_id is not None: normalized_id = str(source_feature_id).strip() if normalized_id: if normalized_id in source_feature_ids: raise AppError( code="DUPLICATE_SOURCE_FEATURE", message=( f"Duplicate source feature {normalized_id} across regional partitions" ), status_code=400, ) source_feature_ids.add(normalized_id) try: geometry = shape(feature.get("geometry")) except Exception as exc: raise AppError( code="GEOMETRY_PARSE_FAILED", message=f"Feature {index} in {partition_path.name} has invalid GeoJSON geometry", status_code=400, ) from exc if geometry.is_empty: raise AppError( code="GEOMETRY_EMPTY", message=f"Feature {index} in {partition_path.name} has an empty geometry", status_code=400, ) if not geometry.is_valid: raise AppError( code="GEOMETRY_INVALID", message=( f"Feature {index} in {partition_path.name} is invalid; " "partitioned ingestion never silently repairs geometry" ), status_code=400, ) _validate_vector_feature_source_schema( feature=feature, properties=properties, geometry=geometry, schema=self._source_schema, feature_context=f"Feature {index} in {partition_path.name}", ) feature_bounds = geometry.bounds if not all(isfinite(value) for value in feature_bounds): raise AppError( code="GEOMETRY_BOUNDS_INVALID", message=f"Feature {index} in {partition_path.name} has non-finite bounds", status_code=400, ) geometry_types.add(geometry.geom_type) min_x = feature_bounds[0] if min_x is None else min(min_x, feature_bounds[0]) min_y = feature_bounds[1] if min_y is None else min(min_y, feature_bounds[1]) max_x = feature_bounds[2] if max_x is None else max(max_x, feature_bounds[2]) max_y = feature_bounds[3] if max_y is None else max(max_y, feature_bounds[3]) feature_count += 1 if feature_count != self._expected_feature_count: raise AppError( code="PARTITION_FEATURE_COUNT_MISMATCH", message=( f"Regional artifact declares {self._expected_feature_count} features but " f"partitions contain {feature_count} features" ), status_code=400, ) if None in {min_x, min_y, max_x, max_y}: # pragma: no cover - feature-count invariant above raise AppError( code="VECTOR_FEATURES_REQUIRED", message="Partitioned vector artifact has no geometry records.", status_code=400, ) bounds_json = { "min_x": float(min_x), "min_y": float(min_y), "max_x": float(max_x), "max_y": float(max_y), } # A MultiPoint envelope is validation evidence only, not a replacement # for persisted source features. It gives the generic contract the # audited aggregate bounds without retaining all Shapely objects. representative_geometry = MultiPoint( [ (bounds_json["min_x"], bounds_json["min_y"]), (bounds_json["max_x"], bounds_json["min_y"]), (bounds_json["max_x"], bounds_json["max_y"]), (bounds_json["min_x"], bounds_json["max_y"]), ] ) return _PartitionedVectorAudit( feature_count=feature_count, geometry_types=tuple(sorted(geometry_types)), bounds_json=bounds_json, partition_checksums_sha256=dict(sorted(observed_checksums.items())), source_schema_validation=self._source_schema.to_metadata(checked_feature_count=feature_count), representative_record=GeometryRecord( geometry=representative_geometry, properties={"partitioned_geometry_audit": True}, identifier="partitioned-geometry-audit", ), ) def _validated_declared_checksums(self) -> dict[str, str]: """Require an exact filename-to-SHA256 manifest for every partition. A list of checksum values is insufficient: it cannot establish which municipality/source partition produced which persisted feature set. The explicit map is also retained with the aggregate audit evidence. """ if not isinstance(self._declared_checksums, dict) or not self._declared_checksums: raise AppError( code="PARTITION_CHECKSUM_MANIFEST_REQUIRED", message="Partitioned ingestion requires a non-empty filename-to-checksum manifest.", status_code=400, ) partition_names = [path.name for path in self._partition_paths] if len(set(partition_names)) != len(partition_names): raise AppError( code="DUPLICATE_PARTITION_IDENTITY", message="Partitioned ingestion requires unique partition filenames.", status_code=400, ) declared = { str(key): str(value).strip().lower() for key, value in self._declared_checksums.items() } if len(declared) != len(partition_names) or set(declared) != set(partition_names): raise AppError( code="PARTITION_CHECKSUM_MANIFEST_MISMATCH", message="Partition checksum manifest must contain exactly one entry for each partition filename.", details={ "expected_partition_filenames": sorted(partition_names), "declared_partition_filenames": sorted(declared), }, status_code=400, ) invalid = sorted(name for name, checksum in declared.items() if not _CHECKSUM_SHA256.fullmatch(checksum)) if invalid: raise AppError( code="PARTITION_CHECKSUM_INVALID", message="Partition checksum manifest contains a non-SHA256 value.", details={"partition_filenames": invalid}, status_code=400, ) return declared class DatasetService: VECTOR_EXTENSIONS = {".geojson", ".json"} RASTER_EXTENSIONS = {".tif", ".tiff", ".geotiff"} VECTOR_TYPES = {"vector", "geojson"} RASTER_TYPES = {"raster", "tif", "tiff", "geotiff"} VALID_DATASET_ROLES = {"source", "derived", "reference"} VALID_TEMPORAL_GRANULARITIES = {"snapshot", "day", "month", "year", "period"} CANONICAL_VECTOR_CRS = "EPSG:4326" @staticmethod def _registry_persistence_available(db: Session) -> bool: """Return true only for real ORM-backed ingestion transactions. Production request handling always supplies a SQLAlchemy Session. The narrow fallback keeps historical lightweight unit fakes (which predate the registry tables) isolated; it cannot bypass the database-backed production import path. """ return callable(getattr(db, "query", None)) @staticmethod def _stable_hash(payload: Any) -> str: return sha256( json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") ).hexdigest() @classmethod def _canonical_vector_storage_bytes(cls, payload: dict[str, Any]) -> bytes: """Serialize the consumable GeoJSON representation deterministically. ``VectorFeatureService.canonicalize_geojson_payload`` is the one place that transforms source coordinates to EPSG:4326. This helper makes the exact result of that transform the persisted, checksummed dataset artifact too; it must never remain merely an in-memory view. """ return json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") @classmethod def _vector_storage_requires_canonicalization(cls, source_crs: str | None) -> bool: """Return whether the source file cannot itself be the canonical view. A missing CRS is intentionally treated as the GeoJSON/RFC-7946 default EPSG:4326. Other aliases (for example ``CRS:84``) are rewritten so every transformed consumption artifact explicitly says ``EPSG:4326``. """ return str(source_crs or cls.CANONICAL_VECTOR_CRS).strip().upper() != cls.CANONICAL_VECTOR_CRS @classmethod def _persist_vector_source_evidence( cls, *, project_id: UUID, dataset_id: UUID, original_filename: str, content: bytes, content_type: str | None, ) -> dict[str, Any]: """Retain a non-canonical source file outside the consumption path. The Dataset's normal ``storage_path`` always points at the canonical artifact. The source bytes are retained only below ``provenance/`` and are referenced through structured provenance metadata; consumers must never treat this location as a dataset input. """ safe_filename = StorageService._safe_filename(original_filename) evidence_path = ( StorageService.dataset_root(str(project_id), str(dataset_id), "vector") / "provenance" / f"{dataset_id}_source_{safe_filename}" ) return StorageService.persist_file( str(evidence_path), content, original_filename=safe_filename, content_type=content_type, ) @classmethod def _record_vector_source_evidence( cls, *, source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], source_crs: str, evidence: dict[str, Any], canonical_checksum_sha256: str, ) -> None: """Bind original source bytes to their canonical consumption artifact.""" source_artifact = { "storage_path": evidence["storage_path"], "checksum_sha256": evidence["checksum_sha256"], "size_bytes": evidence["size_bytes"], "content_type": evidence["content_type"], "source_crs": source_crs, "retention": "provenance_evidence_only", } transformation = { "name": "vector_crs_normalization", "version": "1.0.0", "source_crs": source_crs, "storage_crs": cls.CANONICAL_VECTOR_CRS, "source_checksum_sha256": evidence["checksum_sha256"], "canonical_checksum_sha256": canonical_checksum_sha256, } source_metadata["source_artifact"] = source_artifact provenance_metadata["source_artifact"] = source_artifact provenance_metadata["canonical_consumption_artifact"] = { "checksum_sha256": canonical_checksum_sha256, "crs": cls.CANONICAL_VECTOR_CRS, "storage_role": "dataset_consumption", } provenance_metadata["transformations"] = [ *( provenance_metadata.get("transformations") if isinstance(provenance_metadata.get("transformations"), list) else [] ), transformation, ] @staticmethod def _calculate_file_checksum_sha256(path: str | Path) -> str: """Stream an operator artifact before storage for an idempotent ingest key.""" artifact = Path(path) if not artifact.is_file(): raise AppError( code="DATASET_FILE_MISSING", message="Partitioned vector artifact is missing", details={"artifact_path": str(artifact)}, status_code=404, ) digest = sha256() with artifact.open("rb") as stream: for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() @classmethod def _ingest_key( cls, *, project_id: UUID, source_key: str, checksum_sha256: str, dataset_type: str, dataset_role: str, area_id: UUID | None, reference_layer_name: str | None, source_version: str | None, ) -> str: return cls._stable_hash( { "project_id": str(project_id), "source_key": source_key, "checksum_sha256": checksum_sha256.lower(), "dataset_type": dataset_type, "dataset_role": dataset_role, "area_id": str(area_id) if area_id else None, "reference_layer_name": reference_layer_name or None, "source_version": source_version or None, "ingest_contract": "phase2-source-provenance-v1", } ) @staticmethod def _contract_metadata( *, metadata: dict[str, Any], source_metadata: dict[str, Any] | None, provenance_metadata: dict[str, Any] | None, source: Any, ) -> dict[str, Any]: result = dict(metadata) source_values = source_metadata if isinstance(source_metadata, dict) else {} provenance_values = provenance_metadata if isinstance(provenance_metadata, dict) else {} result["license"] = ( result.get("license") or source_values.get("license") or source_values.get("license_note") or provenance_values.get("license") or getattr(source, "license_name", None) or "unknown" ) result.setdefault( "usage_restrictions", source_values.get("usage_restrictions") or getattr(source, "usage_restrictions", None) or "unknown", ) return result @staticmethod def _validate_vector_source_schema(source: Any, feature_collection: dict[str, Any]) -> dict[str, Any]: """Validate the server-owned source schema after canonicalization. Generic GeoJSON validation proves that a feature collection is structurally valid. This additional pass proves that it also matches the geometry and required-attribute expectations recorded for the selected source registry entry. It deliberately uses the canonical-storage payload so the evidence describes exactly what will be persisted in ``vector_features``. """ schema = _SourceVectorSchema.from_source(source) features = feature_collection.get("features") if isinstance(feature_collection, dict) else None if not isinstance(features, list): raise AppError( code="SOURCE_SCHEMA_FEATURE_COLLECTION_INVALID", message="Source-schema validation requires a GeoJSON FeatureCollection.", status_code=400, ) for index, feature in enumerate(features): if not isinstance(feature, dict): raise AppError( code="SOURCE_SCHEMA_FEATURE_INVALID", message=f"Feature {index} is not an object during source-schema validation.", status_code=400, ) properties = feature.get("properties") if isinstance(feature.get("properties"), dict) else {} try: geometry = shape(feature.get("geometry")) except Exception as exc: raise AppError( code="SOURCE_SCHEMA_GEOMETRY_INVALID", message=f"Feature {index} has no parseable geometry during source-schema validation.", status_code=400, ) from exc _validate_vector_feature_source_schema( feature=feature, properties=properties, geometry=geometry, schema=schema, feature_context=f"Feature {index}", ) return schema.to_metadata(checked_feature_count=len(features)) @staticmethod def _snapshot_freshness_status( source_key: str, source_metadata: dict[str, Any] | None, *, observed_at: datetime | None, source_version: str | None, ) -> str: metadata = source_metadata if isinstance(source_metadata, dict) else {} supplied = str(metadata.get("freshness_status") or "").strip().lower() allowed = {"unknown", "current", "due", "stale", "not_applicable", "review_required"} if supplied in allowed: return supplied if source_key in {"manual", "fixture", "map_selection", "derived", "experimental"}: return "not_applicable" return "current" if observed_at is not None or bool((source_version or "").strip()) else "review_required" @staticmethod def _resolution_unit_for_crs(crs: str | None) -> str: normalized = str(crs or "").strip().upper() return "degree" if normalized in {"EPSG:4326", "CRS:84", "OGC:CRS84"} else "m" @staticmethod def _failed_validation_report( *, asset_id: str, dataset_type: str, code: str, message: str, now: datetime, category: str = "parser", ) -> ValidationReport: if dataset_type == "vector": contract_key, contract_version = VECTOR_GEOJSON_CONTRACT_KEY, VECTOR_GEOJSON_CONTRACT_VERSION else: contract_key, contract_version = RASTER_GEOTIFF_CONTRACT_KEY, RASTER_GEOTIFF_CONTRACT_VERSION return ValidationReport( asset_id=asset_id, data_contract_key=contract_key, data_contract_version=contract_version, contract_fingerprint_sha256=None, validation_status=ValidationStatus.FAILED, provenance_status=ProvenanceStatus.INCOMPLETE, lineage_status=LineageStatus.INCOMPLETE, quarantine_status=QuarantineStatus.QUARANTINED, validation_scope=("ingest", dataset_type), checked_at=now, issues=( ValidationIssue( code=code, category=category, field="artifact", message=message, ), ), ) @staticmethod def _normalize_datetime(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) @staticmethod def _validate_temporal_metadata( *, temporal_series_key: str | None, observed_at: datetime | None, valid_from: datetime | None, valid_to: datetime | None, temporal_granularity: str | None, source_version: str | None, ) -> dict[str, Any]: normalized_key = (temporal_series_key or "").strip() or None normalized_observed_at = DatasetService._normalize_datetime(observed_at) normalized_valid_from = DatasetService._normalize_datetime(valid_from) normalized_valid_to = DatasetService._normalize_datetime(valid_to) normalized_granularity = (temporal_granularity or "").strip().lower() or None normalized_source_version = (source_version or "").strip() or None if normalized_key and len(normalized_key) > 255: raise AppError(code="INVALID_TEMPORAL_METADATA", message="temporal_series_key is too long", status_code=400) if normalized_granularity and normalized_granularity not in DatasetService.VALID_TEMPORAL_GRANULARITIES: raise AppError( code="INVALID_TEMPORAL_METADATA", message="temporal_granularity must be snapshot, day, month, year or period", status_code=400, ) if normalized_valid_from and normalized_valid_to and normalized_valid_to < normalized_valid_from: raise AppError( code="INVALID_TEMPORAL_METADATA", message="valid_to must be on or after valid_from", status_code=400, ) if normalized_key and normalized_observed_at is None: raise AppError( code="INVALID_TEMPORAL_METADATA", message="observed_at is required when temporal_series_key is provided", status_code=400, ) if normalized_observed_at and normalized_key is None: raise AppError( code="INVALID_TEMPORAL_METADATA", message="temporal_series_key is required when observed_at is provided", status_code=400, ) return { "temporal_series_key": normalized_key, "observed_at": normalized_observed_at, "valid_from": normalized_valid_from, "valid_to": normalized_valid_to, "temporal_granularity": normalized_granularity, "source_version": normalized_source_version, } @staticmethod def _to_response(dataset: Dataset) -> DatasetCreateResponse: metadata_json = dataset.metadata_json if isinstance(dataset.metadata_json, dict) else {} return DatasetCreateResponse( id=dataset.id, name=dataset.name, dataset_type=dataset.dataset_type, source=dataset.source, dataset_role=dataset.dataset_role, source_name=dataset.source_name, reference_layer_name=dataset.reference_layer_name, source_metadata=dataset.source_metadata, provenance_metadata=dataset.provenance_metadata, ingest_key=dataset.ingest_key, source_registry_id=dataset.source_registry_id, source_snapshot_id=dataset.source_snapshot_id, data_contract_key=dataset.data_contract_key, data_contract_version=dataset.data_contract_version, validation_status=dataset.validation_status, validation_report_json=dataset.validation_report_json, provenance_status=dataset.provenance_status, lineage_status=dataset.lineage_status, quarantine_status=dataset.quarantine_status, imported_at=dataset.imported_at, temporal_series_key=dataset.temporal_series_key, observed_at=dataset.observed_at, valid_from=dataset.valid_from, valid_to=dataset.valid_to, temporal_granularity=dataset.temporal_granularity, source_version=dataset.source_version, project_id=dataset.project_id, area_id=dataset.area_id, storage_path=dataset.storage_path, original_filename=dataset.original_filename, stored_filename=dataset.stored_filename, content_type=dataset.content_type, size_bytes=dataset.size_bytes, checksum_sha256=dataset.checksum_sha256, crs=dataset.crs, bounds_json=dataset.bounds_json, metadata_json=dataset.metadata_json, vector_summary=DatasetService._extract_vector_summary(dataset.dataset_type, metadata_json), status=dataset.status, derived_from_dataset_id=dataset.derived_from_dataset_id, created_at=dataset.created_at, feature_count=metadata_json.get("feature_count"), ) @staticmethod def _canonical_dataset_type(dataset_type: str) -> str: normalized = (dataset_type or "").strip().lower() if normalized in DatasetService.VECTOR_TYPES: return "vector" if normalized in DatasetService.RASTER_TYPES: return "raster" raise AppError( code="INVALID_DATASET_TYPE", message="dataset_type must be 'vector' or 'raster' (or legacy 'geojson')", status_code=400, ) @staticmethod def _normalize_stored_dataset_type(dataset_type: str) -> str: normalized = (dataset_type or "").strip().lower() if normalized in DatasetService.VECTOR_TYPES: return "vector" if normalized in DatasetService.RASTER_TYPES: return "raster" return normalized @staticmethod def _is_vector_type(dataset_type: str) -> bool: return DatasetService._normalize_stored_dataset_type(dataset_type) == "vector" @staticmethod def _is_raster_type(dataset_type: str) -> bool: return DatasetService._normalize_stored_dataset_type(dataset_type) == "raster" @staticmethod def _normalize_dataset_role(dataset_role: str | None) -> str: normalized = (dataset_role or "").strip().lower() or "source" if normalized not in DatasetService.VALID_DATASET_ROLES: raise AppError( code="INVALID_DATASET_ROLE", message="dataset_role must be one of: source, derived, reference", status_code=400, ) return normalized @staticmethod def _extension_for_path(filename: str) -> str: return Path(filename).suffix.lower() @staticmethod def _validate_upload_filename(filename: str | None) -> str: if not filename: raise AppError(code="INVALID_UPLOAD", message="Missing file name", status_code=400) return filename @staticmethod def list_datasets(db: Session, project_id: UUID, limit: int = 50, offset: int = 0) -> tuple[list[DatasetCreateResponse], int]: total = db.query(Dataset).filter(Dataset.project_id == project_id).count() rows = ( db.query(Dataset) .filter(Dataset.project_id == project_id) .order_by(Dataset.created_at.desc()) .offset(offset) .limit(limit) .all() ) return [DatasetService._to_response(row) for row in rows], total @staticmethod def _extract_vector_summary(dataset_type: str, metadata_json: dict) -> DatasetVectorSummary | None: if not DatasetService._is_vector_type(dataset_type): return None if not isinstance(metadata_json, dict): return None return DatasetVectorSummary( feature_count=metadata_json.get("feature_count"), geometry_types=metadata_json.get("geometry_types"), bounds_json=metadata_json.get("bounds_json"), approximate_area_m2=metadata_json.get("approximate_area_m2"), crs=metadata_json.get("crs"), feature_geometry_count=metadata_json.get("feature_geometry_count"), invalid_features=metadata_json.get("invalid_features"), crs_assumed=metadata_json.get("crs_assumed"), ) @staticmethod def _extract_raster_bounds_json(metadata_json: dict[str, Any]) -> dict[str, float] | None: existing = metadata_json.get("bounds_json") if isinstance(existing, dict): return existing bounds = metadata_json.get("bounds") if isinstance(bounds, (list, tuple)) and len(bounds) == 4: return { "minx": float(bounds[0]), "miny": float(bounds[1]), "maxx": float(bounds[2]), "maxy": float(bounds[3]), } return None @staticmethod def _extract_raster_resolution_json(metadata_json: dict[str, Any]) -> dict[str, float] | None: existing = metadata_json.get("resolution_json") if isinstance(existing, dict): return existing resolution = metadata_json.get("resolution") if isinstance(resolution, (list, tuple)) and len(resolution) >= 2: return {"x": float(resolution[0]), "y": float(resolution[1])} return None @staticmethod def _extract_raster_bands_json(metadata_json: dict[str, Any]) -> dict[str, Any] | None: existing = metadata_json.get("bands_json") if isinstance(existing, dict): return existing bands_json: dict[str, Any] = {} if metadata_json.get("band_count") is not None: bands_json["band_count"] = int(metadata_json["band_count"]) if metadata_json.get("dtype") is not None: bands_json["dtype"] = metadata_json["dtype"] return bands_json or None @classmethod def _find_existing_ingest(cls, db: Session, project_id: UUID, ingest_key: str) -> Dataset | None: if not cls._registry_persistence_available(db): return None return SourceRegistryService.find_dataset_by_ingest_key(db, project_id, ingest_key) @classmethod def _record_snapshot( cls, db: Session, *, source_key: str, checksum_sha256: str, source_version: str | None, observed_at: datetime | None, valid_from: datetime | None, valid_to: datetime | None, source_crs: str | None, source_metadata: dict[str, Any] | None, metadata: dict[str, Any], ) -> tuple[Any | None, Any | None]: if not cls._registry_persistence_available(db): return None, None source = SourceRegistryService.ensure_server_owned_source(db, source_key) source_values = source_metadata if isinstance(source_metadata, dict) else {} resolution = metadata.get("resolution_json") or metadata.get("resolution") or {} if isinstance(resolution, (list, tuple)) and len(resolution) >= 2: resolution = {"x": resolution[0], "y": resolution[1], "unit": cls._resolution_unit_for_crs(source_crs)} if not isinstance(resolution, dict): resolution = {"status": "unknown"} snapshot_key = f"{source_key}:{source_version or 'unversioned'}:{checksum_sha256.lower()}" snapshot = SourceRegistryService.record_snapshot( db, source_key=source_key, snapshot_key=snapshot_key, checksum_sha256=checksum_sha256, source_version=source_version, snapshot_at=observed_at, fetched_at=datetime.now(timezone.utc), reuse_existing_snapshot=True, source_url=( source_values.get("source_url") or source_values.get("catalogue_url") or source_values.get("service_url") ), crs=source_crs, units=source_values.get("units") or source.default_units, spatial_resolution=resolution, temporal_coverage={ "observed_at": observed_at.isoformat() if observed_at else None, "valid_from": valid_from.isoformat() if valid_from else None, "valid_to": valid_to.isoformat() if valid_to else None, }, geographic_coverage={ "bbox": metadata.get("source_bounds_json") or metadata.get("bounds_json") or metadata.get("bounds"), "coverage_zones": source_values.get("coverage_zones") or source_values.get("coverage_zone"), }, observed_schema={ "dataset_type": metadata.get("dataset_type"), "geometry_types": metadata.get("geometry_types"), "bands": metadata.get("band_count"), "attributes": source_values.get("expected_attributes"), }, freshness_status=cls._snapshot_freshness_status( source_key, source_metadata, observed_at=observed_at, source_version=source_version, ), ingest_status="ingested", known_limitations=list(source_values.get("known_limitations") or []), snapshot_metadata={ "source_metadata": source_values, "source_checksum_sha256": checksum_sha256.lower(), }, ) return source, snapshot @classmethod def _apply_validation_report( cls, db: Session, *, dataset: Dataset, dataset_version: DatasetVersion, report: ValidationReport, source: Any | None, snapshot: Any | None, artifact_path: str | None, ) -> None: fields = report.persistence_fields() dataset.validation_report_json = fields["validation_report_json"] dataset.quarantine_status = fields["quarantine_status"] dataset_version.validation_report_json = fields["validation_report_json"] if source is not None and snapshot is not None: SourceRegistryService.bind_dataset_provenance( dataset, source=source, snapshot=snapshot, data_contract_key=fields["data_contract_key"], data_contract_version=fields["data_contract_version"], validation_status=fields["validation_status"], provenance_status=fields["provenance_status"], lineage_status=fields["lineage_status"], ) SourceRegistryService.bind_dataset_version_provenance( dataset_version, source=source, snapshot=snapshot, data_contract_key=fields["data_contract_key"], data_contract_version=fields["data_contract_version"], validation_status=fields["validation_status"], provenance_status=fields["provenance_status"], lineage_status=fields["lineage_status"], ) else: for target in (dataset, dataset_version): target.data_contract_key = fields["data_contract_key"] target.data_contract_version = fields["data_contract_version"] target.validation_status = fields["validation_status"] target.provenance_status = fields["provenance_status"] target.lineage_status = fields["lineage_status"] decision = DataQuarantineService.decide(report) if decision.eligible_for_use: dataset.status = "ready" dataset.quarantine_status = "not_quarantined" return dataset.status = "quarantined" dataset.quarantine_status = "quarantined" if source is not None and snapshot is not None: SourceRegistryService.quarantine_dataset( db, dataset=dataset, dataset_version=dataset_version, source_snapshot=snapshot, stage="ingest_validation", reason_code=(decision.reason_codes[0] if decision.reason_codes else "DATA_CONTRACT_FAILED"), details={"validation_report": report.to_dict(), "quarantine_decision": decision.to_dict()}, artifact_path=artifact_path, artifact_checksum_sha256=dataset.checksum_sha256, ) @classmethod def _new_dataset_version( cls, dataset: Dataset, *, ingest_key: str | None, ) -> DatasetVersion: return DatasetVersion( id=uuid.uuid4(), dataset_id=dataset.id, version=1, storage_path=dataset.storage_path, source_version=dataset.source_version, observed_at=dataset.observed_at, valid_from=dataset.valid_from, valid_to=dataset.valid_to, checksum_sha256=dataset.checksum_sha256, ingest_key=f"{ingest_key}:v1" if ingest_key else None, source_metadata=dataset.source_metadata, provenance_metadata=dataset.provenance_metadata, ) @staticmethod async def _upload_dataset_legacy( db: Session, project_id: UUID, file: UploadFile, dataset_type: str, source: str, dataset_role: str = "source", source_name: str | None = None, reference_layer_name: str | None = None, source_metadata: dict | None = None, provenance_metadata: dict | None = None, area_id: UUID | None = None, temporal_series_key: str | None = None, observed_at: datetime | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, temporal_granularity: str | None = None, source_version: str | None = None, ) -> DatasetCreateResponse: if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) filename = DatasetService._validate_upload_filename(file.filename) canonical_type = DatasetService._canonical_dataset_type(dataset_type) normalized_role = DatasetService._normalize_dataset_role(dataset_role) temporal = DatasetService._validate_temporal_metadata( temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, ) normalized_source_name = source_name if normalized_role == "reference" and not normalized_source_name: normalized_source_name = "manual" if normalized_role == "reference" and canonical_type == "raster": raise AppError(code="INVALID_DATASET_ROLE", message="reference role requires vector dataset type", status_code=400) extension = DatasetService._extension_for_path(filename) if canonical_type == "vector" and extension not in DatasetService.VECTOR_EXTENSIONS: raise AppError(code="INVALID_UPLOAD", message="Vector uploads require .geojson or .json files", status_code=415) if canonical_type == "raster" and extension not in DatasetService.RASTER_EXTENSIONS: raise AppError( code="INVALID_UPLOAD", message="Raster uploads require .tif, .tiff or .geotiff files", status_code=415, ) raw = await file.read() storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id := uuid.uuid4()), dataset_type=canonical_type, original_filename=filename, content=raw, content_type=file.content_type, ) metadata: dict[str, Any] = {} vector_payload: dict[str, Any] | None = None status = "uploaded" try: status = "validating" if canonical_type == "vector": try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: raise AppError(code="INVALID_UPLOAD", message="Upload must be UTF-8 encoded", status_code=400) from exc metadata = parse_geojson_payload(text) vector_payload = json.loads(text) status = "ready" else: metadata = extract_raster_metadata(storage_info["storage_path"]) status = "ready" except ValueError as exc: status = "failed" StorageService.remove_dataset_file(storage_info["storage_path"]) raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc except AppError as exc: if canonical_type == "raster" and exc.code == "RASTER_PROCESSING_UNAVAILABLE": status = "failed" metadata = { "processing_error": exc.message, "processing_code": exc.code, } else: StorageService.remove_dataset_file(storage_info["storage_path"]) raise bounds_json = metadata.get("bounds_json") if isinstance(metadata, dict) else None resolution_json = metadata.get("resolution_json") if isinstance(metadata, dict) else None bands_json = metadata.get("bands_json") if isinstance(metadata, dict) else None if canonical_type == "raster" and isinstance(metadata, dict): bounds_json = DatasetService._extract_raster_bounds_json(metadata) resolution_json = DatasetService._extract_raster_resolution_json(metadata) bands_json = DatasetService._extract_raster_bands_json(metadata) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=filename, dataset_type=canonical_type, source=source, dataset_role=normalized_role, source_name=normalized_source_name, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_metadata=source_metadata, provenance_metadata=provenance_metadata, imported_at=datetime.now(timezone.utc), **temporal, storage_path=storage_info["storage_path"], original_filename=storage_info["original_filename"], stored_filename=storage_info["stored_filename"], content_type=storage_info["content_type"], size_bytes=storage_info["size_bytes"], checksum_sha256=storage_info["checksum_sha256"], crs=metadata.get("crs") if isinstance(metadata, dict) else None, bounds_json=bounds_json, resolution_json=resolution_json, bands_json=bands_json, metadata_json=metadata, status=status, ) try: db.add(dataset) db.add( DatasetVersion( dataset_id=dataset.id, version=1, storage_path=dataset.storage_path, source_version=dataset.source_version, observed_at=dataset.observed_at, valid_from=dataset.valid_from, valid_to=dataset.valid_to, checksum_sha256=dataset.checksum_sha256, source_metadata=dataset.source_metadata, provenance_metadata=dataset.provenance_metadata, ) ) if canonical_type == "vector" and vector_payload is not None and status == "ready": feature_class = reference_layer_name if normalized_role == "reference" else None VectorFeatureService.persist_geojson_features( db=db, dataset_id=dataset.id, payload=vector_payload, feature_class=feature_class, commit=False, ) db.commit() db.refresh(dataset) except Exception: db.rollback() StorageService.remove_dataset_file(storage_info["storage_path"]) raise return DatasetService._to_response(dataset) @staticmethod async def upload_dataset( db: Session, project_id: UUID, file: UploadFile, dataset_type: str, source: str, dataset_role: str = "source", source_name: str | None = None, reference_layer_name: str | None = None, source_metadata: dict | None = None, provenance_metadata: dict | None = None, area_id: UUID | None = None, temporal_series_key: str | None = None, observed_at: datetime | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, temporal_granularity: str | None = None, source_version: str | None = None, ) -> DatasetCreateResponse: """Stage a user upload as an explicitly manual, non-authoritative source. Client text such as ``source_name=grb`` is retained only as a claim in provenance. It cannot select an authoritative registry entry; only a server-owned acquisition adapter reaches those entries. """ if not DatasetService._registry_persistence_available(db): return await DatasetService._upload_dataset_legacy( db=db, project_id=project_id, file=file, dataset_type=dataset_type, source=source, dataset_role=dataset_role, source_name=source_name, reference_layer_name=reference_layer_name, source_metadata=source_metadata, provenance_metadata=provenance_metadata, area_id=area_id, temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, ) if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) if area_id is not None: area = db.get(Area, area_id) if not area: raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) if area.project_id != project_id: raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400) filename = DatasetService._validate_upload_filename(file.filename) canonical_type = DatasetService._canonical_dataset_type(dataset_type) normalized_role = DatasetService._normalize_dataset_role(dataset_role) if normalized_role == "reference" and canonical_type == "raster": raise AppError(code="INVALID_DATASET_ROLE", message="reference role requires vector dataset type", status_code=400) extension = DatasetService._extension_for_path(filename) if canonical_type == "vector" and extension not in DatasetService.VECTOR_EXTENSIONS: raise AppError(code="INVALID_UPLOAD", message="Vector uploads require .geojson or .json files", status_code=415) if canonical_type == "raster" and extension not in DatasetService.RASTER_EXTENSIONS: raise AppError(code="INVALID_UPLOAD", message="Raster uploads require .tif, .tiff or .geotiff files", status_code=415) temporal = DatasetService._validate_temporal_metadata( temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, ) raw = await file.read() checksum_sha256 = StorageService.calculate_checksum_sha256(raw) ingest_key = DatasetService._ingest_key( project_id=project_id, source_key="manual", checksum_sha256=checksum_sha256, dataset_type=canonical_type, dataset_role=normalized_role, area_id=area_id, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_version=temporal["source_version"], ) existing = DatasetService._find_existing_ingest(db, project_id, ingest_key) if existing is not None: return DatasetService._to_response(existing) raw_source_metadata = dict(source_metadata or {}) raw_provenance_metadata = dict(provenance_metadata or {}) raw_source_metadata.update( { "ingest_origin": "manual_upload", "claimed_source": source, "claimed_source_name": source_name, "authority_claim_accepted": False, } ) raw_source_metadata.setdefault( "temporal_unknown_reason", "The manual upload does not assert a precise source observation timestamp.", ) raw_source_metadata.setdefault( "source_version_unknown_reason", "The manual upload has no server-attested source edition or snapshot version.", ) raw_provenance_metadata.update( { "ingest_origin": "manual_upload", "ingest_key": ingest_key, "claimed_source": {"source": source, "source_name": source_name}, } ) dataset_id = uuid.uuid4() storage_info: dict[str, Any] | None = None storage_content = raw source_evidence: dict[str, Any] | None = None imported_at = datetime.now(timezone.utc) metadata: dict[str, Any] = {"dataset_type": canonical_type} source_crs: str | None = None canonical_vector_payload: dict[str, Any] | None = None parser_error: tuple[str, str] | None = None try: if canonical_type == "vector": try: payload = json.loads(raw.decode("utf-8")) except UnicodeDecodeError as exc: raise AppError(code="INVALID_UPLOAD", message="Upload must be UTF-8 encoded", status_code=400) from exc raw_metadata = parse_geojson_payload(payload) source_crs = str(raw_metadata.get("crs") or "").strip() or None canonical_vector_payload = VectorFeatureService.canonicalize_geojson_payload( payload, source_crs=source_crs or DatasetService.CANONICAL_VECTOR_CRS, ) metadata = parse_geojson_payload(canonical_vector_payload) metadata.update( { "dataset_type": "vector", "source_crs": source_crs, "source_bounds_json": raw_metadata.get("bounds_json"), "source_crs_assumed": raw_metadata.get("crs_assumed", False), "canonical_storage_crs": DatasetService.CANONICAL_VECTOR_CRS, } ) if DatasetService._vector_storage_requires_canonicalization(source_crs): storage_content = DatasetService._canonical_vector_storage_bytes(canonical_vector_payload) source_evidence = DatasetService._persist_vector_source_evidence( project_id=project_id, dataset_id=dataset_id, original_filename=filename, content=raw, content_type=file.content_type, ) else: storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type=canonical_type, original_filename=filename, content=raw, content_type=file.content_type, ) metadata = extract_raster_metadata(storage_info["storage_path"]) metadata["dataset_type"] = "raster" source_crs = metadata.get("crs") except (ValueError, json.JSONDecodeError, AppError) as exc: code = exc.code if isinstance(exc, AppError) else "INVALID_GEOJSON" parser_error = (code, str(exc)) metadata = { "dataset_type": canonical_type, "processing_error": str(exc), "processing_code": code, } if storage_info is None: storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type=canonical_type, original_filename=filename, content=storage_content, content_type=file.content_type, ) computed_storage_checksum_sha256 = StorageService.calculate_checksum_sha256(storage_content) if source_evidence is not None: resolved_source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS DatasetService._record_vector_source_evidence( source_metadata=raw_source_metadata, provenance_metadata=raw_provenance_metadata, source_crs=resolved_source_crs, evidence=source_evidence, canonical_checksum_sha256=computed_storage_checksum_sha256, ) metadata.update( { "source_artifact_checksum_sha256": source_evidence["checksum_sha256"], "canonical_artifact_checksum_sha256": computed_storage_checksum_sha256, } ) source_registry, source_snapshot = DatasetService._record_snapshot( db, source_key="manual", checksum_sha256=storage_info["checksum_sha256"], source_version=temporal["source_version"], observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], source_crs=source_crs, source_metadata=raw_source_metadata, metadata=metadata, ) contract_metadata = DatasetService._contract_metadata( metadata=metadata, source_metadata=raw_source_metadata, provenance_metadata=raw_provenance_metadata, source=source_registry, ) if parser_error is not None: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type=canonical_type, code=parser_error[0], message=parser_error[1], now=imported_at, ) elif canonical_type == "vector": source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS lineage = LineageEvidence() if source_crs.upper() != DatasetService.CANONICAL_VECTOR_CRS: lineage = LineageEvidence( transformations=( TransformationEvidence( name="vector_crs_normalization", version="1.0.0", checksum_sha256=DatasetService._stable_hash( {"source_crs": source_crs, "storage_crs": DatasetService.CANONICAL_VECTOR_CRS} ), ), ) ) try: contract_metadata["source_schema_validation"] = DatasetService._validate_vector_source_schema( source_registry, canonical_vector_payload or {"type": "FeatureCollection", "features": []}, ) report = validate_registered_asset( build_vector_ingest_input( asset_id=ingest_key, source_crs=source_crs, storage_crs=DatasetService.CANONICAL_VECTOR_CRS, feature_collection=canonical_vector_payload or {"type": "FeatureCollection", "features": []}, checksum_sha256=storage_info["checksum_sha256"], computed_checksum_sha256=computed_storage_checksum_sha256, content=storage_content, source_registry_id=str(source_registry.id), source_snapshot_id=str(source_snapshot.id), imported_at=imported_at, metadata=contract_metadata, observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], temporal_unknown_reason=raw_source_metadata["temporal_unknown_reason"], source_version=temporal["source_version"], source_version_unknown_reason=raw_source_metadata["source_version_unknown_reason"], lineage=lineage, ) ) except AppError as exc: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type="vector", code=exc.code, message=exc.message, now=imported_at, category="source_schema", ) else: resolution_json = DatasetService._extract_raster_resolution_json(metadata) resolution = ( {"x": resolution_json["x"], "y": resolution_json["y"], "unit": DatasetService._resolution_unit_for_crs(source_crs)} if resolution_json else None ) report = validate_registered_asset( build_raster_ingest_input( asset_id=ingest_key, source_crs=source_crs, storage_crs=source_crs, raster_profile=metadata, bounds=DatasetService._extract_raster_bounds_json(metadata), resolution=resolution, checksum_sha256=storage_info["checksum_sha256"], computed_checksum_sha256=checksum_sha256, content=raw, source_registry_id=str(source_registry.id), source_snapshot_id=str(source_snapshot.id), imported_at=imported_at, metadata=contract_metadata, observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], temporal_unknown_reason=raw_source_metadata["temporal_unknown_reason"], source_version=temporal["source_version"], source_version_unknown_reason=raw_source_metadata["source_version_unknown_reason"], ) ) bounds_json = metadata.get("bounds_json") resolution_json = metadata.get("resolution_json") bands_json = metadata.get("bands_json") if canonical_type == "raster": bounds_json = DatasetService._extract_raster_bounds_json(metadata) resolution_json = DatasetService._extract_raster_resolution_json(metadata) bands_json = DatasetService._extract_raster_bands_json(metadata) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=filename, dataset_type=canonical_type, source="manual_upload", dataset_role=normalized_role, source_name="manual", reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_metadata=raw_source_metadata, provenance_metadata=raw_provenance_metadata, imported_at=imported_at, ingest_key=ingest_key, **temporal, storage_path=storage_info["storage_path"], original_filename=storage_info["original_filename"], stored_filename=storage_info["stored_filename"], content_type=storage_info["content_type"], size_bytes=storage_info["size_bytes"], checksum_sha256=storage_info["checksum_sha256"], crs=(DatasetService.CANONICAL_VECTOR_CRS if canonical_type == "vector" else source_crs), bounds_json=bounds_json, resolution_json=resolution_json, bands_json=bands_json, metadata_json=contract_metadata, status="validating", ) dataset_version = DatasetService._new_dataset_version(dataset, ingest_key=ingest_key) try: db.add(dataset) db.add(dataset_version) db.flush() DatasetService._apply_validation_report( db, dataset=dataset, dataset_version=dataset_version, report=report, source=source_registry, snapshot=source_snapshot, artifact_path=storage_info["storage_path"], ) if report.validation_status == ValidationStatus.PASSED and canonical_vector_payload is not None: VectorFeatureService.persist_geojson_features( db=db, dataset_id=dataset.id, payload=canonical_vector_payload, feature_class=reference_layer_name if normalized_role == "reference" else None, source_crs=DatasetService.CANONICAL_VECTOR_CRS, commit=False, ) db.commit() db.refresh(dataset) except Exception: db.rollback() # Keep the staged bytes. A transport/database failure must remain # inspectable instead of silently deleting the only evidence. raise return DatasetService._to_response(dataset) @staticmethod def _import_vector_bytes_legacy( db: Session, *, project_id: UUID, filename: str, content: bytes, source: str, source_name: str, dataset_role: str, reference_layer_name: str | None, source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], area_id: UUID | None = None, temporal_series_key: str | None = None, observed_at: datetime | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, temporal_granularity: str | None = None, source_version: str | None = None, content_type: str = "application/geo+json", ) -> DatasetCreateResponse: if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) if area_id is not None: area = db.get(Area, area_id) if not area: raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) if area.project_id != project_id: raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400) if not content: raise AppError(code="INVALID_UPLOAD", message="Vector artifact is empty", status_code=400) safe_filename = DatasetService._validate_upload_filename(filename) if DatasetService._extension_for_path(safe_filename) not in DatasetService.VECTOR_EXTENSIONS: raise AppError(code="INVALID_UPLOAD", message="Vector artifacts require .geojson or .json files", status_code=415) normalized_role = DatasetService._normalize_dataset_role(dataset_role) normalized_source_name = (source_name or "").strip() or ("manual" if normalized_role == "reference" else None) temporal = DatasetService._validate_temporal_metadata( temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, ) try: text = content.decode("utf-8") except UnicodeDecodeError as exc: raise AppError(code="INVALID_UPLOAD", message="Vector artifact must be UTF-8 encoded", status_code=400) from exc try: metadata = parse_geojson_payload(text) vector_payload = json.loads(text) except (ValueError, json.JSONDecodeError) as exc: raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc dataset_id = uuid.uuid4() storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type="vector", original_filename=safe_filename, content=content, content_type=content_type, ) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=safe_filename, dataset_type="vector", source=source, dataset_role=normalized_role, source_name=normalized_source_name, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_metadata=source_metadata, provenance_metadata=provenance_metadata, imported_at=datetime.now(timezone.utc), **temporal, storage_path=storage_info["storage_path"], original_filename=storage_info["original_filename"], stored_filename=storage_info["stored_filename"], content_type=storage_info["content_type"], size_bytes=storage_info["size_bytes"], checksum_sha256=storage_info["checksum_sha256"], crs=metadata.get("crs"), bounds_json=metadata.get("bounds_json"), metadata_json=metadata, status="ready", ) try: db.add(dataset) db.add( DatasetVersion( dataset_id=dataset.id, version=1, storage_path=dataset.storage_path, source_version=dataset.source_version, observed_at=dataset.observed_at, valid_from=dataset.valid_from, valid_to=dataset.valid_to, checksum_sha256=dataset.checksum_sha256, source_metadata=dataset.source_metadata, provenance_metadata=dataset.provenance_metadata, ) ) VectorFeatureService.persist_geojson_features( db=db, dataset_id=dataset.id, payload=vector_payload, feature_class=reference_layer_name if normalized_role == "reference" else None, commit=False, ) db.commit() db.refresh(dataset) except Exception: db.rollback() StorageService.remove_dataset_file(storage_info["storage_path"]) raise return DatasetService._to_response(dataset) @staticmethod def _import_raster_bytes_legacy( db: Session, *, project_id: UUID, filename: str, content: bytes, source: str, source_name: str, source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], area_id: UUID | None = None, temporal_series_key: str | None = None, observed_at: datetime | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, temporal_granularity: str | None = None, source_version: str | None = None, content_type: str = "image/tiff", ) -> DatasetCreateResponse: if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) if area_id is not None: area = db.get(Area, area_id) if not area: raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) if area.project_id != project_id: raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400) if not content: raise AppError(code="INVALID_UPLOAD", message="Raster artifact is empty", status_code=400) safe_filename = DatasetService._validate_upload_filename(filename) if DatasetService._extension_for_path(safe_filename) not in DatasetService.RASTER_EXTENSIONS: raise AppError(code="INVALID_UPLOAD", message="Raster artifacts require a GeoTIFF filename", status_code=415) temporal = DatasetService._validate_temporal_metadata( temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, ) dataset_id = uuid.uuid4() storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type="raster", original_filename=safe_filename, content=content, content_type=content_type, ) try: metadata = extract_raster_metadata(storage_info["storage_path"]) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=safe_filename, dataset_type="raster", source=source, dataset_role="source", source_name=source_name, source_metadata=source_metadata, provenance_metadata=provenance_metadata, imported_at=datetime.now(timezone.utc), **temporal, storage_path=storage_info["storage_path"], original_filename=storage_info["original_filename"], stored_filename=storage_info["stored_filename"], content_type=storage_info["content_type"], size_bytes=storage_info["size_bytes"], checksum_sha256=storage_info["checksum_sha256"], crs=metadata.get("crs"), bounds_json=DatasetService._extract_raster_bounds_json(metadata), resolution_json=DatasetService._extract_raster_resolution_json(metadata), bands_json=DatasetService._extract_raster_bands_json(metadata), metadata_json=metadata, status="ready", ) db.add(dataset) db.add( DatasetVersion( dataset_id=dataset.id, version=1, storage_path=dataset.storage_path, source_version=dataset.source_version, observed_at=dataset.observed_at, valid_from=dataset.valid_from, valid_to=dataset.valid_to, checksum_sha256=dataset.checksum_sha256, source_metadata=dataset.source_metadata, provenance_metadata=dataset.provenance_metadata, ) ) db.commit() db.refresh(dataset) return DatasetService._to_response(dataset) except Exception: db.rollback() StorageService.remove_dataset_file(storage_info["storage_path"]) raise @staticmethod def _governed_import_bytes( db: Session, *, project_id: UUID, filename: str, content: bytes, dataset_type: str, source: str, source_name: str, dataset_role: str, reference_layer_name: str | None, source_metadata: dict[str, Any] | None, provenance_metadata: dict[str, Any] | None, area_id: UUID | None, temporal_series_key: str | None, observed_at: datetime | None, valid_from: datetime | None, valid_to: datetime | None, temporal_granularity: str | None, source_version: str | None, content_type: str, ) -> DatasetCreateResponse: """Persist an adapter-owned source through one governed ingestion path. Acquisition adapters choose an entry from the server-owned registry; they cannot create authority identities dynamically. The original artifact is deliberately retained if parsing or validation fails so the immutable checksum, source snapshot and quarantine record remain reviewable. """ if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) if area_id is not None: area = db.get(Area, area_id) if not area: raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) if area.project_id != project_id: raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400) if not content: raise AppError(code="INVALID_UPLOAD", message=f"{dataset_type.title()} artifact is empty", status_code=400) canonical_type = DatasetService._canonical_dataset_type(dataset_type) normalized_role = DatasetService._normalize_dataset_role(dataset_role) if normalized_role == "reference" and canonical_type == "raster": raise AppError(code="INVALID_DATASET_ROLE", message="reference role requires vector dataset type", status_code=400) safe_filename = DatasetService._validate_upload_filename(filename) extension = DatasetService._extension_for_path(safe_filename) allowed_extensions = DatasetService.VECTOR_EXTENSIONS if canonical_type == "vector" else DatasetService.RASTER_EXTENSIONS if extension not in allowed_extensions: expected = ".geojson or .json" if canonical_type == "vector" else ".tif, .tiff or .geotiff" raise AppError(code="INVALID_UPLOAD", message=f"{canonical_type.title()} artifacts require {expected} files", status_code=415) source_key = SourceRegistryService.normalize_source_key(source_name) # This lookup deliberately happens before writing the artifact. A # typo in an internal adapter must not acquire an unregistered source # identity or silently downgrade itself to a manual source. SourceRegistryService.definition_for(source_key) temporal = DatasetService._validate_temporal_metadata( temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, ) computed_checksum_sha256 = StorageService.calculate_checksum_sha256(content) ingest_key = DatasetService._ingest_key( project_id=project_id, source_key=source_key, checksum_sha256=computed_checksum_sha256, dataset_type=canonical_type, dataset_role=normalized_role, area_id=area_id, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_version=temporal["source_version"], ) existing = DatasetService._find_existing_ingest(db, project_id, ingest_key) if existing is not None: return DatasetService._to_response(existing) governed_source_metadata = dict(source_metadata or {}) governed_provenance_metadata = dict(provenance_metadata or {}) governed_source_metadata.update( { "ingest_origin": "governed_acquisition_adapter", "source_registry_key": source_key, "authority_claim_accepted": True, } ) governed_source_metadata.setdefault( "temporal_unknown_reason", "The governed source did not publish a precise observation timestamp for this snapshot.", ) governed_source_metadata.setdefault( "source_version_unknown_reason", "The governed source did not publish a stable source edition; the immutable checksum identifies this snapshot.", ) governed_provenance_metadata.update( { "ingest_origin": "governed_acquisition_adapter", "source_registry_key": source_key, "ingest_key": ingest_key, } ) dataset_id = uuid.uuid4() storage_info: dict[str, Any] | None = None storage_content = content source_evidence: dict[str, Any] | None = None imported_at = datetime.now(timezone.utc) metadata: dict[str, Any] = {"dataset_type": canonical_type} source_crs: str | None = None canonical_vector_payload: dict[str, Any] | None = None parser_error: tuple[str, str] | None = None try: if canonical_type == "vector": try: payload = json.loads(content.decode("utf-8")) except UnicodeDecodeError as exc: raise AppError(code="INVALID_UPLOAD", message="Vector artifact must be UTF-8 encoded", status_code=400) from exc raw_metadata = parse_geojson_payload(payload) source_crs = str(raw_metadata.get("crs") or "").strip() or None canonical_vector_payload = VectorFeatureService.canonicalize_geojson_payload( payload, source_crs=source_crs or DatasetService.CANONICAL_VECTOR_CRS, ) metadata = parse_geojson_payload(canonical_vector_payload) metadata.update( { "dataset_type": "vector", "source_crs": source_crs, "source_bounds_json": raw_metadata.get("bounds_json"), "source_crs_assumed": raw_metadata.get("crs_assumed", False), "canonical_storage_crs": DatasetService.CANONICAL_VECTOR_CRS, } ) if DatasetService._vector_storage_requires_canonicalization(source_crs): storage_content = DatasetService._canonical_vector_storage_bytes(canonical_vector_payload) source_evidence = DatasetService._persist_vector_source_evidence( project_id=project_id, dataset_id=dataset_id, original_filename=safe_filename, content=content, content_type=content_type, ) else: storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type=canonical_type, original_filename=safe_filename, content=content, content_type=content_type, ) metadata = extract_raster_metadata(storage_info["storage_path"]) metadata["dataset_type"] = "raster" source_crs = str(metadata.get("crs") or "").strip() or None except (ValueError, json.JSONDecodeError, AppError) as exc: code = exc.code if isinstance(exc, AppError) else "INVALID_GEOJSON" parser_error = (code, str(exc)) metadata = {"dataset_type": canonical_type, "processing_error": str(exc), "processing_code": code} if storage_info is None: storage_info = StorageService.persist_dataset_file( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type=canonical_type, original_filename=safe_filename, content=storage_content, content_type=content_type, ) computed_storage_checksum_sha256 = StorageService.calculate_checksum_sha256(storage_content) if source_evidence is not None: resolved_source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS DatasetService._record_vector_source_evidence( source_metadata=governed_source_metadata, provenance_metadata=governed_provenance_metadata, source_crs=resolved_source_crs, evidence=source_evidence, canonical_checksum_sha256=computed_storage_checksum_sha256, ) metadata.update( { "source_artifact_checksum_sha256": source_evidence["checksum_sha256"], "canonical_artifact_checksum_sha256": computed_storage_checksum_sha256, } ) source_registry, source_snapshot = DatasetService._record_snapshot( db, source_key=source_key, checksum_sha256=storage_info["checksum_sha256"], source_version=temporal["source_version"], observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], source_crs=source_crs, source_metadata=governed_source_metadata, metadata=metadata, ) contract_metadata = DatasetService._contract_metadata( metadata=metadata, source_metadata=governed_source_metadata, provenance_metadata=governed_provenance_metadata, source=source_registry, ) if parser_error is not None: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type=canonical_type, code=parser_error[0], message=parser_error[1], now=imported_at, ) elif canonical_type == "vector": resolved_source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS lineage = LineageEvidence() if resolved_source_crs.upper() != DatasetService.CANONICAL_VECTOR_CRS: lineage = LineageEvidence( transformations=( TransformationEvidence( name="vector_crs_normalization", version="1.0.0", checksum_sha256=DatasetService._stable_hash( {"source_crs": resolved_source_crs, "storage_crs": DatasetService.CANONICAL_VECTOR_CRS} ), ), ) ) try: contract_metadata["source_schema_validation"] = DatasetService._validate_vector_source_schema( source_registry, canonical_vector_payload or {"type": "FeatureCollection", "features": []}, ) report = validate_registered_asset( build_vector_ingest_input( asset_id=ingest_key, source_crs=resolved_source_crs, storage_crs=DatasetService.CANONICAL_VECTOR_CRS, feature_collection=canonical_vector_payload or {"type": "FeatureCollection", "features": []}, checksum_sha256=storage_info["checksum_sha256"], computed_checksum_sha256=computed_storage_checksum_sha256, content=storage_content, source_registry_id=str(source_registry.id), source_snapshot_id=str(source_snapshot.id), imported_at=imported_at, metadata=contract_metadata, observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], temporal_unknown_reason=governed_source_metadata["temporal_unknown_reason"], source_version=temporal["source_version"], source_version_unknown_reason=governed_source_metadata["source_version_unknown_reason"], lineage=lineage, ) ) except AppError as exc: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type="vector", code=exc.code, message=exc.message, now=imported_at, category="source_schema", ) else: resolution_json = DatasetService._extract_raster_resolution_json(metadata) resolution = ( { "x": resolution_json["x"], "y": resolution_json["y"], "unit": DatasetService._resolution_unit_for_crs(source_crs), } if resolution_json else None ) report = validate_registered_asset( build_raster_ingest_input( asset_id=ingest_key, source_crs=source_crs, storage_crs=source_crs, raster_profile=metadata, bounds=DatasetService._extract_raster_bounds_json(metadata), resolution=resolution, checksum_sha256=storage_info["checksum_sha256"], computed_checksum_sha256=computed_checksum_sha256, content=content, source_registry_id=str(source_registry.id), source_snapshot_id=str(source_snapshot.id), imported_at=imported_at, metadata=contract_metadata, observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], temporal_unknown_reason=governed_source_metadata["temporal_unknown_reason"], source_version=temporal["source_version"], source_version_unknown_reason=governed_source_metadata["source_version_unknown_reason"], ) ) bounds_json = metadata.get("bounds_json") resolution_json = metadata.get("resolution_json") bands_json = metadata.get("bands_json") if canonical_type == "raster": bounds_json = DatasetService._extract_raster_bounds_json(metadata) resolution_json = DatasetService._extract_raster_resolution_json(metadata) bands_json = DatasetService._extract_raster_bands_json(metadata) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=safe_filename, dataset_type=canonical_type, source=source, dataset_role=normalized_role, source_name=source_key, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_metadata=governed_source_metadata, provenance_metadata=governed_provenance_metadata, imported_at=imported_at, ingest_key=ingest_key, **temporal, storage_path=storage_info["storage_path"], original_filename=storage_info["original_filename"], stored_filename=storage_info["stored_filename"], content_type=storage_info["content_type"], size_bytes=storage_info["size_bytes"], checksum_sha256=storage_info["checksum_sha256"], crs=(DatasetService.CANONICAL_VECTOR_CRS if canonical_type == "vector" else source_crs), bounds_json=bounds_json, resolution_json=resolution_json, bands_json=bands_json, metadata_json=contract_metadata, status="validating", ) dataset_version = DatasetService._new_dataset_version(dataset, ingest_key=ingest_key) try: db.add(dataset) db.add(dataset_version) db.flush() DatasetService._apply_validation_report( db, dataset=dataset, dataset_version=dataset_version, report=report, source=source_registry, snapshot=source_snapshot, artifact_path=storage_info["storage_path"], ) if report.validation_status == ValidationStatus.PASSED and canonical_vector_payload is not None: VectorFeatureService.persist_geojson_features( db=db, dataset_id=dataset.id, payload=canonical_vector_payload, feature_class=reference_layer_name if normalized_role == "reference" else None, source_crs=DatasetService.CANONICAL_VECTOR_CRS, commit=False, ) db.commit() db.refresh(dataset) except Exception: db.rollback() # Leave the staged artifact untouched. A failed persistence # transaction is not evidence that the source bytes were safe to # delete or that an acquisition can be repeated silently. raise return DatasetService._to_response(dataset) @staticmethod def import_vector_bytes( db: Session, *, project_id: UUID, filename: str, content: bytes, source: str, source_name: str, dataset_role: str, reference_layer_name: str | None, source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], area_id: UUID | None = None, temporal_series_key: str | None = None, observed_at: datetime | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, temporal_granularity: str | None = None, source_version: str | None = None, content_type: str = "application/geo+json", ) -> DatasetCreateResponse: if not DatasetService._registry_persistence_available(db): return DatasetService._import_vector_bytes_legacy( db, project_id=project_id, filename=filename, content=content, source=source, source_name=source_name, dataset_role=dataset_role, reference_layer_name=reference_layer_name, source_metadata=source_metadata, provenance_metadata=provenance_metadata, area_id=area_id, temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, content_type=content_type, ) return DatasetService._governed_import_bytes( db, project_id=project_id, filename=filename, content=content, dataset_type="vector", source=source, source_name=source_name, dataset_role=dataset_role, reference_layer_name=reference_layer_name, source_metadata=source_metadata, provenance_metadata=provenance_metadata, area_id=area_id, temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, content_type=content_type, ) @staticmethod def import_raster_bytes( db: Session, *, project_id: UUID, filename: str, content: bytes, source: str, source_name: str, source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], area_id: UUID | None = None, temporal_series_key: str | None = None, observed_at: datetime | None = None, valid_from: datetime | None = None, valid_to: datetime | None = None, temporal_granularity: str | None = None, source_version: str | None = None, content_type: str = "image/tiff", ) -> DatasetCreateResponse: if not DatasetService._registry_persistence_available(db): return DatasetService._import_raster_bytes_legacy( db, project_id=project_id, filename=filename, content=content, source=source, source_name=source_name, source_metadata=source_metadata, provenance_metadata=provenance_metadata, area_id=area_id, temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, content_type=content_type, ) return DatasetService._governed_import_bytes( db, project_id=project_id, filename=filename, content=content, dataset_type="raster", source=source, source_name=source_name, dataset_role="source", reference_layer_name=None, source_metadata=source_metadata, provenance_metadata=provenance_metadata, area_id=area_id, temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=valid_from, valid_to=valid_to, temporal_granularity=temporal_granularity, source_version=source_version, content_type=content_type, ) @staticmethod def import_partitioned_vector_artifact( db: Session, *, project_id: UUID, area_id: UUID, artifact_path: str | Path, partition_paths: list[str | Path], original_filename: str, source: str, dataset_role: str, source_name: str, reference_layer_name: str | None, metadata_json: dict[str, Any], source_metadata: dict[str, Any], provenance_metadata: dict[str, Any], temporal_series_key: str, observed_at: datetime, temporal_granularity: str = "snapshot", source_version: str | None = None, batch_size: int = 1000, ) -> DatasetCreateResponse: if not DatasetService._registry_persistence_available(db): raise AppError( code="SOURCE_REGISTRY_PERSISTENCE_UNAVAILABLE", message="Partitioned authoritative imports require registry and provenance persistence.", status_code=503, ) if not db.get(Project, project_id): raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404) area = db.get(Area, area_id) if not area: raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404) if area.project_id != project_id: raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400) if not partition_paths: raise AppError( code="INVALID_GEOJSON_PARTITIONS", message="At least one GeoJSON partition is required", status_code=400, ) filename = DatasetService._validate_upload_filename(original_filename) if DatasetService._extension_for_path(filename) not in DatasetService.VECTOR_EXTENSIONS: raise AppError(code="INVALID_UPLOAD", message="Vector artifacts require .geojson or .json files", status_code=415) normalized_role = DatasetService._normalize_dataset_role(dataset_role) source_key = SourceRegistryService.normalize_source_key(source_name) # A partitioned operator artifact is never allowed to manufacture a # source identity from its caller-provided label. SourceRegistryService.definition_for(source_key) temporal = DatasetService._validate_temporal_metadata( temporal_series_key=temporal_series_key, observed_at=observed_at, valid_from=observed_at, valid_to=None, temporal_granularity=temporal_granularity, source_version=source_version, ) metadata = dict(metadata_json) expected_feature_count = int(metadata.get("feature_count") or 0) if expected_feature_count <= 0: raise AppError( code="INVALID_GEOJSON_PARTITIONS", message="Partition metadata must declare a positive feature_count", status_code=400, ) artifact_checksum_sha256 = DatasetService._calculate_file_checksum_sha256(artifact_path) ingest_key = DatasetService._ingest_key( project_id=project_id, source_key=source_key, checksum_sha256=artifact_checksum_sha256, dataset_type="vector", dataset_role=normalized_role, area_id=area_id, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_version=temporal["source_version"], ) existing = DatasetService._find_existing_ingest(db, project_id, ingest_key) if existing is not None: return DatasetService._to_response(existing) governed_source_metadata = dict(source_metadata or {}) governed_provenance_metadata = dict(provenance_metadata or {}) governed_source_metadata.update( { "ingest_origin": "governed_partitioned_acquisition_adapter", "source_registry_key": source_key, "authority_claim_accepted": True, "partitioned_artifact": True, } ) governed_source_metadata.setdefault( "temporal_unknown_reason", "The governed source did not publish a precise observation timestamp for this snapshot.", ) governed_source_metadata.setdefault( "source_version_unknown_reason", "The governed source did not publish a stable source edition; the immutable checksum identifies this snapshot.", ) governed_provenance_metadata.update( { "ingest_origin": "governed_partitioned_acquisition_adapter", "source_registry_key": source_key, "ingest_key": ingest_key, "artifact_checksum_sha256": artifact_checksum_sha256, "combined_artifact_checksum_sha256": artifact_checksum_sha256, "partition_count": len(partition_paths), } ) dataset_id = uuid.uuid4() storage_info = StorageService.persist_dataset_file_from_path( project_id=str(project_id), dataset_id=str(dataset_id), dataset_type="vector", original_filename=filename, source_path=artifact_path, content_type="application/geo+json", ) # The source file may have changed while it was copied. Re-key on the # bytes actually retained; never bind a snapshot to a stale pre-copy # checksum. persisted_checksum_sha256 = str(storage_info["checksum_sha256"]) if persisted_checksum_sha256 != artifact_checksum_sha256: ingest_key = DatasetService._ingest_key( project_id=project_id, source_key=source_key, checksum_sha256=persisted_checksum_sha256, dataset_type="vector", dataset_role=normalized_role, area_id=area_id, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_version=temporal["source_version"], ) existing = DatasetService._find_existing_ingest(db, project_id, ingest_key) if existing is not None: StorageService.remove_dataset_file(str(storage_info["storage_path"])) return DatasetService._to_response(existing) governed_provenance_metadata["ingest_key"] = ingest_key governed_provenance_metadata["artifact_checksum_sha256"] = persisted_checksum_sha256 governed_provenance_metadata["combined_artifact_checksum_sha256"] = persisted_checksum_sha256 storage_crs = str( metadata.get("canonical_storage_crs") or metadata.get("storage_crs") or metadata.get("crs") or DatasetService.CANONICAL_VECTOR_CRS ).strip() source_crs = str(metadata.get("source_crs") or storage_crs).strip() or None metadata.update( { "dataset_type": "vector", "canonical_storage_crs": DatasetService.CANONICAL_VECTOR_CRS, "partitioned_artifact": True, "partition_count": len(partition_paths), } ) source_registry, source_snapshot = DatasetService._record_snapshot( db, source_key=source_key, checksum_sha256=persisted_checksum_sha256, source_version=temporal["source_version"], observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], source_crs=source_crs, source_metadata=governed_source_metadata, metadata=metadata, ) # Registry persistence is checked above, so absence here is an # infrastructure fault rather than a state that can be imported. if source_registry is None or source_snapshot is None: # pragma: no cover - defensive invariant raise AppError( code="SOURCE_REGISTRY_PERSISTENCE_UNAVAILABLE", message="Source registry persistence did not return a governed snapshot.", status_code=503, ) contract_metadata = DatasetService._contract_metadata( metadata=metadata, source_metadata=governed_source_metadata, provenance_metadata=governed_provenance_metadata, source=source_registry, ) partition_records = _PartitionedGeoJsonRecords( partition_paths, expected_feature_count=expected_feature_count, declared_partition_checksums=governed_provenance_metadata.get("partition_checksums"), source_schema=_SourceVectorSchema.from_source(source_registry), ) lineage = LineageEvidence() if source_crs and storage_crs.upper() != source_crs.upper(): lineage = LineageEvidence( transformations=( TransformationEvidence( name="partitioned_vector_crs_normalization", version="1.0.0", checksum_sha256=DatasetService._stable_hash( { "source_crs": source_crs, "storage_crs": storage_crs, "partition_count": len(partition_paths), } ), ), ) ) declared_artifact_checksum = str(governed_provenance_metadata.get("artifact_sha256") or "").strip().lower() artifact_binding_error: tuple[str, str] | None = None if not declared_artifact_checksum: artifact_binding_error = ( "ARTIFACT_CHECKSUM_REQUIRED", "Partitioned ingestion requires the acquisition manifest's combined artifact checksum.", ) elif not _CHECKSUM_SHA256.fullmatch(declared_artifact_checksum): artifact_binding_error = ( "ARTIFACT_CHECKSUM_INVALID", "Declared partitioned artifact checksum must be a lowercase SHA-256 value.", ) elif declared_artifact_checksum != persisted_checksum_sha256: artifact_binding_error = ( "ARTIFACT_CHECKSUM_MISMATCH", "Declared artifact checksum does not match the retained partitioned artifact.", ) try: partition_audit = partition_records.audit() contract_metadata["partitioned_geometry_audit"] = partition_audit.to_metadata() contract_metadata["source_schema_validation"] = partition_audit.source_schema_validation governed_provenance_metadata["partition_checksum_manifest_sha256"] = DatasetService._stable_hash( partition_audit.partition_checksums_sha256 ) governed_provenance_metadata["partitioned_artifact_binding_sha256"] = DatasetService._stable_hash( { "combined_artifact_checksum_sha256": persisted_checksum_sha256, "partition_checksum_manifest_sha256": governed_provenance_metadata[ "partition_checksum_manifest_sha256" ], "feature_count": partition_audit.feature_count, "storage_crs": storage_crs, } ) if artifact_binding_error is not None: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type="vector", code=artifact_binding_error[0], message=artifact_binding_error[1], now=datetime.now(timezone.utc), category="checksum", ) else: report = validate_registered_asset( DataAssetValidationInput( asset_id=ingest_key, data_contract_key=VECTOR_GEOJSON_CONTRACT_KEY, data_contract_version=VECTOR_GEOJSON_CONTRACT_VERSION, kind=ContractKind.VECTOR, source_crs=source_crs, storage_crs=storage_crs, bounds=contract_metadata.get("bounds_json"), checksum_sha256=persisted_checksum_sha256, computed_checksum_sha256=persisted_checksum_sha256, metadata=contract_metadata, # The partition-bounded audit above validates every # source feature. The generic contract receives only # compact aggregate geometry evidence and therefore # cannot materialize the complete regional artifact. geometry_records=(partition_audit.representative_record,), source_registry_id=str(source_registry.id), source_snapshot_id=str(source_snapshot.id), lineage=lineage, imported_at=datetime.now(timezone.utc), observed_at=temporal["observed_at"], valid_from=temporal["valid_from"], valid_to=temporal["valid_to"], temporal_unknown_reason=governed_source_metadata["temporal_unknown_reason"], source_version=temporal["source_version"], source_version_unknown_reason=governed_source_metadata[ "source_version_unknown_reason" ], ) ) except AppError as exc: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type="vector", code=exc.code, message=exc.message, now=datetime.now(timezone.utc), category="source_schema" if exc.code.startswith("SOURCE_SCHEMA") else "parser", ) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=filename, dataset_type="vector", source=source, dataset_role=normalized_role, source_name=source_key, reference_layer_name=reference_layer_name if normalized_role == "reference" else None, source_metadata=governed_source_metadata, provenance_metadata=governed_provenance_metadata, imported_at=datetime.now(timezone.utc), **temporal, ingest_key=ingest_key, storage_path=storage_info["storage_path"], original_filename=storage_info["original_filename"], stored_filename=storage_info["stored_filename"], content_type=storage_info["content_type"], size_bytes=storage_info["size_bytes"], checksum_sha256=persisted_checksum_sha256, crs=storage_crs, bounds_json=metadata.get("bounds_json"), metadata_json=contract_metadata, status="validating", ) dataset_version = DatasetService._new_dataset_version(dataset, ingest_key=ingest_key) try: db.add(dataset) db.add(dataset_version) db.flush() if report.validation_status == ValidationStatus.PASSED: try: begin_nested = getattr(db, "begin_nested", None) if callable(begin_nested): with begin_nested(): persisted_count = VectorFeatureService.persist_geojson_partitions( db, dataset.id, partition_paths, feature_class=reference_layer_name if normalized_role == "reference" else None, batch_size=batch_size, source_crs=storage_crs, ) if persisted_count != expected_feature_count: raise AppError( code="PARTITION_FEATURE_COUNT_MISMATCH", message=( f"Regional artifact declares {expected_feature_count} features but " f"{persisted_count} queryable features were indexed" ), status_code=400, ) else: # lightweight test sessions only; production uses a savepoint persisted_count = VectorFeatureService.persist_geojson_partitions( db, dataset.id, partition_paths, feature_class=reference_layer_name if normalized_role == "reference" else None, batch_size=batch_size, source_crs=storage_crs, ) if persisted_count != expected_feature_count: raise AppError( code="PARTITION_FEATURE_COUNT_MISMATCH", message=( f"Regional artifact declares {expected_feature_count} features but " f"{persisted_count} queryable features were indexed" ), status_code=400, ) except AppError as exc: report = DatasetService._failed_validation_report( asset_id=ingest_key, dataset_type="vector", code=exc.code, message=exc.message, now=datetime.now(timezone.utc), ) DatasetService._apply_validation_report( db, dataset=dataset, dataset_version=dataset_version, report=report, source=source_registry, snapshot=source_snapshot, artifact_path=str(storage_info["storage_path"]), ) db.commit() db.refresh(dataset) except Exception: db.rollback() # Retain staged bytes for forensic review. A transaction error is # not evidence that the source artifact may be safely discarded. raise return DatasetService._to_response(dataset) @staticmethod def refresh_metadata(db: Session, dataset_id: UUID) -> DatasetCreateResponse: dataset = DatasetService._get_dataset(db, dataset_id) if dataset.quarantine_status == "quarantined" or dataset.status == "quarantined": raise AppError( code="DATASET_QUARANTINED", message="Quarantined datasets cannot be refreshed into an eligible state; re-ingest a new governed snapshot.", status_code=409, ) # A governed dataset's source snapshot and validation report bind the # exact bytes, CRS and extracted metadata that were inspected at # ingest. Re-reading a mutable storage path here would otherwise let # an in-place replacement change the operational artifact while its # persisted checksum/report still says ``passed``. Such a change must # create a new immutable source snapshot through the governed ingest # path; metadata refresh remains intentionally available only to rows # without Phase-2 contract evidence. has_governed_contract_evidence = any( ( dataset.source_registry_id is not None, dataset.source_snapshot_id is not None, bool(str(dataset.data_contract_key or "").strip()), bool(str(dataset.data_contract_version or "").strip()), dataset.validation_report_json is not None, dataset.validation_status == "passed", ) ) if has_governed_contract_evidence: raise AppError( code="GOVERNED_DATASET_REINGEST_REQUIRED", message=( "Governed dataset metadata is immutable evidence. Re-ingest the artifact to create a new " "source snapshot and validation report." ), status_code=409, ) if not dataset.storage_path: raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) if not Path(dataset.storage_path).exists(): raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) try: if DatasetService._is_vector_type(dataset.dataset_type): metadata = parse_geojson_payload(load_dataset_text(dataset.storage_path)) elif DatasetService._is_raster_type(dataset.dataset_type): metadata = extract_raster_metadata(dataset.storage_path) else: raise AppError(code="INVALID_DATASET_TYPE", message="Cannot refresh metadata for this dataset type", status_code=400) except ValueError as exc: raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc except AppError as exc: if DatasetService._is_raster_type(dataset.dataset_type) and exc.code == "RASTER_PROCESSING_UNAVAILABLE": metadata = {"processing_error": exc.message, "processing_code": exc.code} else: raise bounds_json = metadata.get("bounds_json") if isinstance(metadata, dict) else dataset.bounds_json resolution_json = metadata.get("resolution_json") if isinstance(metadata, dict) else dataset.resolution_json bands_json = metadata.get("bands_json") if isinstance(metadata, dict) else dataset.bands_json if DatasetService._is_raster_type(dataset.dataset_type) and isinstance(metadata, dict): bounds_json = DatasetService._extract_raster_bounds_json(metadata) resolution_json = DatasetService._extract_raster_resolution_json(metadata) bands_json = DatasetService._extract_raster_bands_json(metadata) # Metadata extraction is observational only. It must never turn an # unvalidated historical row into a ready, authoritative dataset. if DatasetService._is_vector_type(dataset.dataset_type): dataset.crs = DatasetService.CANONICAL_VECTOR_CRS else: dataset.crs = metadata.get("crs") if isinstance(metadata, dict) else dataset.crs dataset.bounds_json = bounds_json existing_metadata = dataset.metadata_json if isinstance(dataset.metadata_json, dict) else {} dataset.metadata_json = {**existing_metadata, **metadata} dataset.resolution_json = resolution_json dataset.bands_json = bands_json db.add(dataset) db.commit() db.refresh(dataset) return DatasetService._to_response(dataset) @staticmethod def update_temporal_metadata(db: Session, dataset_id: UUID, payload: DatasetTemporalUpdate) -> DatasetCreateResponse: dataset = DatasetService._get_dataset(db, dataset_id) if dataset.quarantine_status == "quarantined" or dataset.status == "quarantined": raise AppError( code="DATASET_QUARANTINED", message="Quarantined datasets require a new governed ingest rather than an in-place temporal edit.", status_code=409, ) temporal = DatasetService._validate_temporal_metadata(**payload.model_dump()) if all(getattr(dataset, field) == value for field, value in temporal.items()): return DatasetService._to_response(dataset) for field, value in temporal.items(): setattr(dataset, field, value) # Observation/source-version fields are contract inputs. Their edit # invalidates the prior report, so later training/inference gates fail # closed until a governed re-ingest persists a new snapshot/report. dataset.status = "validating" dataset.validation_status = "not_validated" dataset.validation_report_json = None dataset.provenance_status = "incomplete" dataset.quarantine_status = "not_quarantined" latest_version = ( db.query(DatasetVersion) .filter(DatasetVersion.dataset_id == dataset.id) .order_by(DatasetVersion.version.desc()) .first() ) db.add(dataset) db.add( DatasetVersion( dataset_id=dataset.id, version=(latest_version.version + 1) if latest_version else 1, storage_path=dataset.storage_path, source_version=dataset.source_version, observed_at=dataset.observed_at, valid_from=dataset.valid_from, valid_to=dataset.valid_to, checksum_sha256=dataset.checksum_sha256, ingest_key=( f"{dataset.ingest_key}:temporal:{(latest_version.version + 1) if latest_version else 1}" if dataset.ingest_key else None ), source_metadata=dataset.source_metadata, provenance_metadata=dataset.provenance_metadata, source_registry_id=dataset.source_registry_id, source_snapshot_id=dataset.source_snapshot_id, data_contract_key=dataset.data_contract_key, data_contract_version=dataset.data_contract_version, validation_status="not_validated", validation_report_json=None, provenance_status="incomplete", lineage_status=dataset.lineage_status, ) ) db.commit() db.refresh(dataset) return DatasetService._to_response(dataset) @staticmethod def list_versions(db: Session, dataset_id: UUID) -> list[DatasetVersionRead]: DatasetService._get_dataset(db, dataset_id) rows = ( db.query(DatasetVersion) .filter(DatasetVersion.dataset_id == dataset_id) .order_by(DatasetVersion.version.desc()) .all() ) return [DatasetVersionRead.model_validate(row) for row in rows] @staticmethod def get_dataset(db: Session, dataset_id: UUID) -> Dataset: dataset = db.get(Dataset, dataset_id) if not dataset: raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404) return dataset @staticmethod def _get_dataset(db: Session, dataset_id: UUID) -> Dataset: return DatasetService.get_dataset(db, dataset_id) @staticmethod def get_dataset_geojson(db: Session, dataset_id: UUID) -> dict: dataset = DatasetService._get_dataset(db, dataset_id) if not DatasetService._is_vector_type(dataset.dataset_type): raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a vector dataset", status_code=400) if not dataset.storage_path: raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) if not pathlib.Path(dataset.storage_path).exists(): raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) raw = load_dataset_text(dataset.storage_path) try: payload = json.loads(raw) metadata_value = getattr(dataset, "metadata_json", None) metadata = metadata_value if isinstance(metadata_value, dict) else {} provenance_value = getattr(dataset, "provenance_metadata", None) provenance = provenance_value if isinstance(provenance_value, dict) else {} canonical_evidence = provenance.get("canonical_consumption_artifact") canonical_checksum = ( canonical_evidence.get("checksum_sha256") if isinstance(canonical_evidence, dict) else metadata.get("canonical_artifact_checksum_sha256") ) # Post-normalization imports persist canonical bytes. Reapplying # their original source CRS here would transform those coordinates # a second time. Historical rows without this immutable binding # retain the legacy read-time canonicalization behavior until they # are re-ingested through the governed path. source_crs = ( DatasetService.CANONICAL_VECTOR_CRS if canonical_checksum == getattr(dataset, "checksum_sha256", None) else ( metadata.get("source_crs") or getattr(dataset, "crs", None) or DatasetService.CANONICAL_VECTOR_CRS ) ) return VectorFeatureService.canonicalize_geojson_payload(payload, source_crs=str(source_crs)) except Exception as exc: raise AppError(code="INVALID_GEOJSON", message="Stored dataset is not valid JSON", status_code=500) from exc @staticmethod def inspect_vector_dataset(db: Session, dataset_id: UUID) -> dict[str, Any]: dataset = DatasetService._get_dataset(db, dataset_id) if not DatasetService._is_vector_type(dataset.dataset_type): raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a vector dataset", status_code=400) if not dataset.storage_path or not Path(dataset.storage_path).exists(): raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) metadata = dataset.metadata_json or {} if not isinstance(metadata, dict): metadata = {} summary = DatasetService._extract_vector_summary(dataset.dataset_type, metadata) return { "dataset": { "id": str(dataset.id), "name": dataset.name, "dataset_type": dataset.dataset_type, "status": dataset.status, "source": dataset.source, "storage": DatasetStorageResponse( original_filename=dataset.original_filename, stored_filename=dataset.stored_filename, content_type=dataset.content_type, size_bytes=dataset.size_bytes, checksum_sha256=dataset.checksum_sha256, ).model_dump(), "feature_count": metadata.get("feature_count"), "crs": metadata.get("crs"), }, "summary": summary.model_dump() if summary else None, "metadata": metadata, } @staticmethod def vector_summary(db: Session, dataset_id: UUID) -> dict[str, Any]: dataset = DatasetService._get_dataset(db, dataset_id) if not DatasetService._is_vector_type(dataset.dataset_type): raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a vector dataset", status_code=400) metadata = dataset.metadata_json or {} if not isinstance(metadata, dict): metadata = {} summary = DatasetService._extract_vector_summary(dataset.dataset_type, metadata) if not summary: raise AppError(code="INVALID_GEOJSON", message="Vector summary unavailable", status_code=422) return summary.model_dump() @staticmethod def raster_metadata(db: Session, dataset_id: UUID) -> dict[str, Any]: dataset = DatasetService._get_dataset(db, dataset_id) if dataset.quarantine_status == "quarantined" or dataset.status == "quarantined": raise AppError( code="DATASET_QUARANTINED", message="Quarantined datasets cannot be read as production-ready raster metadata.", status_code=409, ) if not DatasetService._is_raster_type(dataset.dataset_type): raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a raster dataset", status_code=400) if not dataset.storage_path: raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) if not Path(dataset.storage_path).exists(): raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404) if isinstance(dataset.metadata_json, dict) and dataset.metadata_json.get("driver"): return dataset.metadata_json metadata = extract_raster_metadata(dataset.storage_path) dataset.metadata_json = dict(dataset.metadata_json or {}) dataset.metadata_json.update(metadata) db.add(dataset) db.commit() db.refresh(dataset) return metadata