refactor: Enhance transfer of waterlevelscontinuous_pressure_daily

- Cache Thing IDs and map PointID to thing_id to satisfy new FK
- Filter orphan rows to prevent invalid inserts
- Add focused transfer unit test to validate mapping and filtering

Author: ksmuczynski

tests/transfers/test_waterlevelscontinuous_pressure_daily_transfer.py

@@ -0,0 +1,47 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+
+import pandas as pd
+
+from transfers.waterlevelscontinuous_pressure_daily import (
+    NMA_WaterLevelsContinuous_Pressure_DailyTransferer,
+)
+
+
+def test_pressure_daily_transfer_filters_orphans(water_well_thing):
+    transferer = NMA_WaterLevelsContinuous_Pressure_DailyTransferer(batch_size=1)
+    df = pd.DataFrame(
+        [
+            {"PointID": water_well_thing.name, "GlobalID": "gid-1"},
+            {"PointID": "MISSING-THING", "GlobalID": "gid-2"},
+        ]
+    )
+
+    filtered = transferer._filter_to_valid_things(df)
+
+    assert list(filtered["PointID"]) == [water_well_thing.name]
+
+
+def test_pressure_daily_row_dict_sets_thing_id(water_well_thing):
+    transferer = NMA_WaterLevelsContinuous_Pressure_DailyTransferer(batch_size=1)
+    row = {"PointID": water_well_thing.name, "GlobalID": "gid-3"}
+
+    mapped = transferer._row_dict(row)
+
+    assert mapped["thing_id"] == water_well_thing.id
+
+
+# ============= EOF =============================================

transfers/waterlevelscontinuous_pressure_daily.py

@@ -22,7 +22,8 @@
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.orm import Session
 
-from db import NMA_WaterLevelsContinuous_Pressure_Daily
+from db import NMA_WaterLevelsContinuous_Pressure_Daily, Thing
+from db.engine import session_ctx
 from transfers.logger import logger
 from transfers.transferer import Transferer
 from transfers.util import read_csv
@@ -41,15 +42,39 @@ class NMA_WaterLevelsContinuous_Pressure_DailyTransferer(Transferer):
     def __init__(self, *args, batch_size: int = 1000, **kwargs):
         super().__init__(*args, **kwargs)
         self.batch_size = batch_size
+        self._thing_id_cache: dict[str, int] = {}
+        self._build_thing_id_cache()
+
+    def _build_thing_id_cache(self) -> None:
+        with session_ctx() as session:
+            things = session.query(Thing.name, Thing.id).all()
+            self._thing_id_cache = {name: thing_id for name, thing_id in things}
+        logger.info(f"Built Thing ID cache with {len(self._thing_id_cache)} entries")
+
+    def _filter_to_valid_things(self, df: pd.DataFrame) -> pd.DataFrame:
+        valid_point_ids = set(self._thing_id_cache.keys())
+        before_count = len(df)
+        filtered_df = df[df["PointID"].isin(valid_point_ids)].copy()
+        after_count = len(filtered_df)
+        if before_count > after_count:
+            skipped = before_count - after_count
+            logger.warning(
+                "Filtered out %s WaterLevelsContinuous_Pressure_Daily records without matching Things "
+                "(%s valid, %s orphan records prevented)",
+                skipped,
+                after_count,
+                skipped,
+            )
+        return filtered_df
 
     def _get_dfs(self) -> tuple[pd.DataFrame, pd.DataFrame]:
         # Parse key datetime columns eagerly to avoid per-row parsing later.
         input_df = read_csv(
             self.source_table,
             parse_dates=["DateMeasured", "Created", "Updated"],
         )
-        # No special cleaning/validation beyond raw import; keep identical copy.
-        return input_df, input_df
+        cleaned_df = self._filter_to_valid_things(input_df)
+        return input_df, cleaned_df
 
     def _transfer_hook(self, session: Session) -> None:
         rows = self._dedupe_rows(
@@ -71,6 +96,7 @@ def _transfer_hook(self, session: Session) -> None:
                     "OBJECTID": excluded.OBJECTID,
                     "WellID": excluded.WellID,
                     "PointID": excluded.PointID,
+                    "thing_id": excluded.thing_id,
                     "DateMeasured": excluded.DateMeasured,
                     "TemperatureWater": excluded.TemperatureWater,
                     "WaterHead": excluded.WaterHead,
@@ -104,6 +130,7 @@ def val(key: str) -> Optional[Any]:
             "OBJECTID": val("OBJECTID"),
             "WellID": val("WellID"),
             "PointID": val("PointID"),
+            "thing_id": self._thing_id_cache.get(val("PointID")),
             "DateMeasured": val("DateMeasured"),
             "TemperatureWater": val("TemperatureWater"),
             "WaterHead": val("WaterHead"),