feat(nma_legacy): establish relationship between NMA_SurfaceWaterPhotos and NMA_SurfaceWaterData via SurfaceID

- make NMA_SurfaceWaterData.surface_id unique to support FK target
- add FK + relationship on NMA_SurfaceWaterPhotos.surface_id with validation
- enforce parent-first ordering by running SurfaceWaterPhotos after SurfaceWaterData
- skip orphan photo rows in transfer using cached SurfaceIDs
- add migration to backfill/cleanup and enforce NOT NULL + FK
- update legacy tests to create photos with real parent SurfaceWaterData

Author: ksmuczynski

alembic/versions/a1b2c3d4e5f6_link_surface_water_photos_to_surface_water_data.py

@@ -0,0 +1,72 @@
+"""link surface water photos to surface water data
+
+Revision ID: a1b2c3d4e5f6
+Revises: f6e5d4c3b2a1
+Create Date: 2026-02-05 11:10:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a1b2c3d4e5f6"
+down_revision: Union[str, Sequence[str], None] = "f6e5d4c3b2a1"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_unique_constraint(
+        "uq_surface_water_data_surface_id",
+        "NMA_SurfaceWaterData",
+        ["SurfaceID"],
+    )
+    op.create_foreign_key(
+        "fk_surface_water_photos_surface_id",
+        "NMA_SurfaceWaterPhotos",
+        "NMA_SurfaceWaterData",
+        ["SurfaceID"],
+        ["SurfaceID"],
+        ondelete="CASCADE",
+    )
+    op.execute(
+        """
+        DELETE FROM "NMA_SurfaceWaterPhotos" p
+        WHERE p."SurfaceID" IS NULL
+           OR NOT EXISTS (
+                SELECT 1
+                FROM "NMA_SurfaceWaterData" d
+                WHERE d."SurfaceID" = p."SurfaceID"
+           )
+        """
+    )
+    op.alter_column(
+        "NMA_SurfaceWaterPhotos",
+        "SurfaceID",
+        existing_type=sa.UUID(),
+        nullable=False,
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.alter_column(
+        "NMA_SurfaceWaterPhotos",
+        "SurfaceID",
+        existing_type=sa.UUID(),
+        nullable=True,
+    )
+    op.drop_constraint(
+        "fk_surface_water_photos_surface_id",
+        "NMA_SurfaceWaterPhotos",
+        type_="foreignkey",
+    )
+    op.drop_constraint(
+        "uq_surface_water_data_surface_id",
+        "NMA_SurfaceWaterData",
+        type_="unique",
+    )

db/nma_legacy.py

@@ -586,7 +586,7 @@ class NMA_SurfaceWaterData(Base):
 
     # Legacy PK (for audit)
     surface_id: Mapped[uuid.UUID] = mapped_column(
-        "SurfaceID", UUID(as_uuid=True), nullable=False
+        "SurfaceID", UUID(as_uuid=True), nullable=False, unique=True
     )
 
     # Legacy FK (for audit)
@@ -617,6 +617,12 @@ class NMA_SurfaceWaterData(Base):
 
     # Relationships
     thing: Mapped["Thing"] = relationship("Thing", back_populates="surface_water_data")
+    surface_water_photos: Mapped[list["NMA_SurfaceWaterPhotos"]] = relationship(
+        "NMA_SurfaceWaterPhotos",
+        back_populates="surface_water_data",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+    )
 
     @validates("thing_id")
     def validate_thing_id(self, key, value):
@@ -632,7 +638,7 @@ class NMA_SurfaceWaterPhotos(Base):
     """
     Legacy SurfaceWaterPhotos table from NM_Aquifer.
 
-    Note: This table is OUT OF SCOPE for refactoring (not a Thing child).
+    Note: This table is a child of NMA_SurfaceWaterData via SurfaceID.
     """
 
     __tablename__ = "NMA_SurfaceWaterPhotos"
@@ -643,21 +649,39 @@ class NMA_SurfaceWaterPhotos(Base):
     )
 
     # FK
-    # FK not assigned.
+    surface_id: Mapped[uuid.UUID] = mapped_column(
+        "SurfaceID",
+        UUID(as_uuid=True),
+        ForeignKey("NMA_SurfaceWaterData.SurfaceID", ondelete="CASCADE"),
+        nullable=False,
+    )
 
     # Legacy PK (for audit)
     # Current `global_id` is also the original PK in the legacy DB
 
     # Legacy FK (for audit)
-    surface_id: Mapped[Optional[uuid.UUID]] = mapped_column(
-        "SurfaceID", UUID(as_uuid=True)
-    )
+    # surface_id is also the legacy FK in the source table.
 
     # Additional columns
     point_id: Mapped[str] = mapped_column("PointID", String(50), nullable=False)
     ole_path: Mapped[Optional[str]] = mapped_column("OLEPath", String(50))
     object_id: Mapped[Optional[int]] = mapped_column("OBJECTID", Integer, unique=True)
 
+    # Relationships
+    surface_water_data: Mapped["NMA_SurfaceWaterData"] = relationship(
+        "NMA_SurfaceWaterData", back_populates="surface_water_photos"
+    )
+
+    @validates("surface_id")
+    def validate_surface_id(self, key, value):
+        """Prevent orphan NMA_SurfaceWaterPhotos - must have a parent NMA_SurfaceWaterData."""
+        if value is None:
+            raise ValueError(
+                "NMA_SurfaceWaterPhotos requires a parent NMA_SurfaceWaterData "
+                "(surface_id cannot be None)"
+            )
+        return value
+
 
 class NMA_WeatherData(Base):
     """

tests/test_surface_water_photos_legacy.py

@@ -28,14 +28,43 @@
 from uuid import uuid4
 
 from db.engine import session_ctx
-from db.nma_legacy import NMA_SurfaceWaterPhotos
+from db.nma_legacy import NMA_SurfaceWaterData, NMA_SurfaceWaterPhotos
+from db.thing import Thing
 
 
-def test_create_surface_water_photos_all_fields():
+def _next_object_id() -> int:
+    # Use a negative value to avoid collisions with existing legacy OBJECTIDs.
+    return -(uuid4().int % 2_000_000_000)
+
+
+def _attach_thing_with_location(session, water_well_thing):
+    location_id = uuid4()
+    thing = session.get(Thing, water_well_thing.id)
+    thing.nma_pk_location = str(location_id)
+    session.commit()
+    return thing, location_id
+
+
+def _create_surface_water_data(session, water_well_thing):
+    thing, location_id = _attach_thing_with_location(session, water_well_thing)
+    record = NMA_SurfaceWaterData(
+        location_id=location_id,
+        thing_id=thing.id,
+        surface_id=uuid4(),
+        point_id="SW-1000",
+        object_id=_next_object_id(),
+    )
+    session.add(record)
+    session.commit()
+    return record
+
+
+def test_create_surface_water_photos_all_fields(water_well_thing):
     """Test creating a surface water photos record with all fields."""
     with session_ctx() as session:
+        parent = _create_surface_water_data(session, water_well_thing)
         record = NMA_SurfaceWaterPhotos(
-            surface_id=uuid4(),
+            surface_id=parent.surface_id,
             point_id="SW-0001",
             ole_path="photo.jpg",
             object_id=123,
@@ -52,14 +81,17 @@ def test_create_surface_water_photos_all_fields():
         assert record.object_id == 123
 
         session.delete(record)
+        session.delete(parent)
         session.commit()
 
 
-def test_create_surface_water_photos_minimal():
+def test_create_surface_water_photos_minimal(water_well_thing):
     """Test creating a surface water photos record with required fields only."""
     with session_ctx() as session:
+        parent = _create_surface_water_data(session, water_well_thing)
         record = NMA_SurfaceWaterPhotos(
             point_id="SW-0002",
+            surface_id=parent.surface_id,
             global_id=uuid4(),
         )
         session.add(record)
@@ -68,11 +100,12 @@ def test_create_surface_water_photos_minimal():
 
         assert record.global_id is not None
         assert record.point_id == "SW-0002"
-        assert record.surface_id is None
+        assert record.surface_id is not None
         assert record.ole_path is None
         assert record.object_id is None
 
         session.delete(record)
+        session.delete(parent)
         session.commit()
 
 

transfers/surface_water_photos.py

@@ -23,7 +23,8 @@
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.orm import Session
 
-from db import NMA_SurfaceWaterPhotos
+from db import NMA_SurfaceWaterData, NMA_SurfaceWaterPhotos
+from db.engine import session_ctx
 from transfers.logger import logger
 from transfers.transferer import Transferer
 from transfers.util import replace_nans
@@ -37,19 +38,44 @@ class SurfaceWaterPhotosTransferer(Transferer):
     def __init__(self, *args, batch_size: int = 1000, **kwargs):
         super().__init__(*args, **kwargs)
         self.batch_size = batch_size
+        self._surface_id_cache: set[str] = set()
+        self._build_surface_id_cache()
+
+    def _build_surface_id_cache(self) -> None:
+        with session_ctx() as session:
+            surface_ids = session.query(NMA_SurfaceWaterData.surface_id).all()
+            for (surface_id,) in surface_ids:
+                if surface_id:
+                    self._surface_id_cache.add(self._normalize_surface_id(surface_id))
+        logger.info(
+            "Built SurfaceWaterData cache with %s surface ids",
+            len(self._surface_id_cache),
+        )
 
     def _get_dfs(self) -> tuple[pd.DataFrame, pd.DataFrame]:
         df = self._read_csv(self.source_table)
         cleaned_df = replace_nans(df)
         return df, cleaned_df
 
     def _transfer_hook(self, session: Session) -> None:
-        rows = [self._row_dict(row) for row in self.cleaned_df.to_dict("records")]
+        rows: list[dict[str, Any]] = []
+        skipped_missing_parent = 0
+        for raw in self.cleaned_df.to_dict("records"):
+            record = self._row_dict(raw)
+            if record is None:
+                skipped_missing_parent += 1
+                continue
+            rows.append(record)
         rows = self._dedupe_rows(rows, key="GlobalID")
 
         if not rows:
             logger.info("No SurfaceWaterPhotos rows to transfer")
             return
+        if skipped_missing_parent:
+            logger.warning(
+                "Skipped %s SurfaceWaterPhotos rows without matching SurfaceWaterData",
+                skipped_missing_parent,
+            )
 
         insert_stmt = insert(NMA_SurfaceWaterPhotos)
         excluded = insert_stmt.excluded
@@ -74,9 +100,16 @@ def _transfer_hook(self, session: Session) -> None:
             session.execute(stmt)
         session.commit()
 
-    def _row_dict(self, row: dict[str, Any]) -> dict[str, Any]:
+    def _row_dict(self, row: dict[str, Any]) -> Optional[dict[str, Any]]:
+        surface_id = self._uuid_val(row.get("SurfaceID"))
+        if surface_id is None or not self._has_surface_id(surface_id):
+            logger.warning(
+                "Skipping SurfaceWaterPhotos SurfaceID=%s - SurfaceWaterData not found",
+                surface_id,
+            )
+            return None
         return {
-            "SurfaceID": self._uuid_val(row.get("SurfaceID")),
+            "SurfaceID": surface_id,
             "PointID": row.get("PointID"),
             "OLEPath": row.get("OLEPath"),
             "OBJECTID": row.get("OBJECTID"),
@@ -107,6 +140,13 @@ def _uuid_val(self, value: Any) -> Optional[UUID]:
                 return None
         return None
 
+    def _has_surface_id(self, surface_id: UUID) -> bool:
+        return self._normalize_surface_id(surface_id) in self._surface_id_cache
+
+    @staticmethod
+    def _normalize_surface_id(value: UUID) -> str:
+        return str(value).strip().lower()
+
 
 def run(batch_size: int = 1000) -> None:
     """Entrypoint to execute the transfer."""

transfers/transfer.py

@@ -513,8 +513,6 @@ def _transfer_parallel(
         parallel_tasks_1.append(("LinkIdsLocation", LinkIdsLocationDataTransferer))
     if opts.transfer_groups:
         parallel_tasks_1.append(("Groups", ProjectGroupTransferer))
-    if opts.transfer_surface_water_photos:
-        parallel_tasks_1.append(("SurfaceWaterPhotos", SurfaceWaterPhotosTransferer))
     if opts.transfer_soil_rock_results:
         parallel_tasks_1.append(("SoilRockResults", SoilRockResultsTransferer))
     if opts.transfer_weather_photos:
@@ -599,8 +597,6 @@ def _transfer_parallel(
         metrics.location_link_ids_metrics(*results_map["LinkIdsLocation"])
     if "Groups" in results_map and results_map["Groups"]:
         metrics.group_metrics(*results_map["Groups"])
-    if "SurfaceWaterPhotos" in results_map and results_map["SurfaceWaterPhotos"]:
-        metrics.surface_water_photos_metrics(*results_map["SurfaceWaterPhotos"])
     if "SoilRockResults" in results_map and results_map["SoilRockResults"]:
         metrics.soil_rock_results_metrics(*results_map["SoilRockResults"])
     if "Assets" in results_map and results_map["Assets"]:
@@ -629,6 +625,11 @@ def _transfer_parallel(
     if "WeatherPhotos" in results_map and results_map["WeatherPhotos"]:
         metrics.weather_photos_metrics(*results_map["WeatherPhotos"])
 
+    if opts.transfer_surface_water_photos:
+        message("TRANSFERRING SURFACE WATER PHOTOS")
+        results = _execute_transfer(SurfaceWaterPhotosTransferer, flags=flags)
+        metrics.surface_water_photos_metrics(*results)
+
     if opts.transfer_major_chemistry:
         message("TRANSFERRING MAJOR CHEMISTRY")
         results = _execute_transfer(MajorChemistryTransferer, flags=flags)