Decouple transfer tests from GCS using local CSV fixtures

tests/transfers/test_contact_with_multiple_wells.py

@@ -14,18 +14,28 @@
 # limitations under the License.
 # ===============================================================================
 
+from pathlib import Path
+
 from db import ThingContactAssociation
 from db.engine import session_ctx
 from transfers.contact_transfer import ContactTransfer
 from transfers.well_transfer import WellTransferer
 
 
 def test_multiple_wells():
-    pointids = ["MG-022", "MG-030", "MG-043"]
-    wt = WellTransferer(pointids=pointids)
+    base_dir = Path(__file__).resolve().parents[2]
+    csv_dir = base_dir / "transfers" / "data"
+    csv_paths = {
+        "WellData": csv_dir / "WellData.csv",
+        "Location": csv_dir / "Location.csv",
+        "OwnersData": csv_dir / "OwnersData.csv",
+        "OwnerLink": csv_dir / "OwnerLink.csv",
+    }
+    pointids = ["TV-230", "EB-317", "SA-0313"]
+    wt = WellTransferer(pointids=pointids, flags={"CSV_PATHS": csv_paths})
     wt.transfer()
 
-    ct = ContactTransfer(pointids=pointids)
+    ct = ContactTransfer(pointids=pointids, flags={"CSV_PATHS": csv_paths})
     ct.transfer()
 
     with session_ctx() as sess:

transfers/contact_transfer.py

@@ -16,10 +16,6 @@
 import json
 
 import pandas as pd
-from pandas import DataFrame
-from pydantic import ValidationError
-from sqlalchemy.orm import Session
-
 from core.enums import Organization
 from db import (
     Contact,
@@ -30,12 +26,15 @@
     IncompleteNMAPhone,
     Base,
 )
+from pandas import DataFrame
+from pydantic import ValidationError
+from sqlalchemy.orm import Session
 from transfers.logger import logger
 from transfers.transferer import ThingBasedTransferer
+from transfers.util import filter_to_valid_point_ids, replace_nans
 from transfers.util import (
     get_transfers_data_path,
 )
-from transfers.util import read_csv, filter_to_valid_point_ids, replace_nans
 
 
 class ContactTransfer(ThingBasedTransferer):
@@ -70,11 +69,11 @@ def calculate_missing_organizations(self):
                 logger.critical(f"Invalid Organization {e}")
 
     def _get_dfs(self):
-        input_df = read_csv(self.source_table)
+        input_df = self._read_csv(self.source_table)
         odf = input_df.drop(["OBJECTID", "GlobalID"], axis=1)
-        ldf = read_csv("OwnerLink")
+        ldf = self._read_csv("OwnerLink")
         ldf = ldf.drop(["OBJECTID", "GlobalID"], axis=1)
-        locdf = read_csv("Location")
+        locdf = self._read_csv("Location")
         ldf = ldf.join(locdf.set_index("LocationId"), on="LocationId")
 
         odf = odf.join(ldf.set_index("OwnerKey"), on="OwnerKey")
@@ -150,7 +149,7 @@ def _add_first_contact(session, row, thing, co_to_org_mapper, added):
         email = _make_email(
             "first",
             row.OwnerKey,
-            email=row.Email.strip(),
+            email=str(row.Email).strip() if row.Email is not None else None,
             email_type="Primary",
             release_status=release_status,
         )
@@ -302,12 +301,18 @@ def _make_name(first, last):
         return f"{first} {last}"
 
 
+def _normalize_string_field(payload: dict, field: str) -> None:
+    value = payload.get(field)
+    if value is None:
+        return
+    payload[field] = str(value).strip()
+
+
 def _make_email(first_second, ownerkey, **kw):
     from schemas.contact import CreateEmail
 
     try:
-        if "email" in kw:
-            kw["email"] = kw["email"].strip()
+        _normalize_string_field(kw, "email")
 
         email = CreateEmail(**kw)
         return Email(**email.model_dump())
@@ -321,8 +326,7 @@ def _make_phone(first_second, ownerkey, **kw):
     from schemas.contact import CreatePhone
 
     try:
-        if "phone_number" in kw:
-            kw["phone_number"] = kw["phone_number"].strip()
+        _normalize_string_field(kw, "phone_number")
 
         phone = CreatePhone(**kw)
         return Phone(**phone.model_dump()), True

transfers/transferer.py

@@ -16,15 +16,14 @@
 import time
 
 import pandas as pd
+from db import Thing, Base
+from db.engine import session_ctx
 from pandas import DataFrame
 from pydantic import ValidationError
 from sqlalchemy.exc import DatabaseError
 from sqlalchemy.orm import Session
-
-from db import Thing, Base
-from db.engine import session_ctx
 from transfers.logger import logger
-from transfers.util import chunk_by_size
+from transfers.util import chunk_by_size, read_csv
 
 
 class ManualFixer(object):
@@ -132,6 +131,15 @@ def _after_hook(self, session: Session):
     def _get_dfs(self):
         raise NotImplementedError("Must implement _get_dfs method")
 
+    def _read_csv(self, name: str, dtype: dict | None = None, **kw) -> pd.DataFrame:
+        if dtype is not None and "dtype" not in kw:
+            kw["dtype"] = dtype
+        csv_paths = self.flags.get("CSV_PATHS") or {}
+        csv_path = csv_paths.get(name)
+        if csv_path:
+            return pd.read_csv(csv_path, **kw)
+        return read_csv(name, dtype=dtype, **kw)
+
 
 class ChunkTransferer(Transferer):
     def __init__(self, *args, **kwargs):

transfers/well_transfer.py

@@ -15,18 +15,13 @@
 # ===============================================================================
 import os
 import re
-import time
 import threading
+import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, UTC
 from zoneinfo import ZoneInfo
 
 import pandas as pd
-from pandas import isna, notna
-from pydantic import ValidationError
-from sqlalchemy.exc import DatabaseError
-from sqlalchemy.orm import Session
-
 from core.enums import (
     WellPurpose as WellPurposeEnum,
     CasingMaterial as WellCasingMaterialEnum,
@@ -47,20 +42,23 @@
     GeologicFormation,
     ThingAquiferAssociation,
 )
+from db.engine import session_ctx
+from pandas import isna, notna
+from pydantic import ValidationError
 from schemas.thing import CreateWell, CreateWellScreen
 from services.gcs_helper import get_storage_bucket
 from services.util import (
     get_state_from_point,
     get_county_from_point,
     get_quad_name_from_point,
 )
-from db.engine import session_ctx
+from sqlalchemy.exc import DatabaseError
+from sqlalchemy.orm import Session
 from transfers.transferer import ChunkTransferer, Transferer
 from transfers.util import (
     make_location,
     make_location_data_provenance,
     filter_to_valid_point_ids,
-    read_csv,
     logger,
     replace_nans,
     get_transferable_wells,
@@ -220,14 +218,19 @@ class WellTransferer(Transferer):
 
     def __init__(self, *args, **kw):
         super().__init__(*args, **kw)
-        self._cached_elevations = get_cached_elevations()
+        if self.flags.get("CSV_PATHS"):
+            # Local CSV mode runs in test/dev without GCS; disable the shared cache
+            # so elevation lookups don’t reach out to the bucket.
+            self._cached_elevations = {}
+        else:
+            self._cached_elevations = get_cached_elevations()
         self._added_locations = {}
         self._aquifers = None
         self._measuring_point_estimator = MeasuringPointEstimator()
 
     def _get_dfs(self):
-        wdf = read_csv("WellData", dtype={"OSEWelltagID": str})
-        ldf = read_csv("Location")
+        wdf = self._read_csv("WellData", dtype={"OSEWelltagID": str})
+        ldf = self._read_csv("Location")
         ldf = ldf.drop(["PointID", "SSMA_TimeStamp"], axis=1)
         wdf = wdf.join(ldf.set_index("LocationId"), on="LocationId")
         wdf = wdf[wdf["SiteType"] == "GW"]
@@ -622,7 +625,10 @@ def _get_or_create_aquifer_system(
             return None, False
 
     def _after_hook(self, session):
-        dump_cached_elevations(self._cached_elevations)
+        if self.flags.get("CSV_PATHS"):
+            logger.info("Skipping cached elevation dump while using CSV_PATHS")
+        else:
+            dump_cached_elevations(self._cached_elevations)
 
         self._row_by_pointid = {
             pid: row
@@ -1592,7 +1598,7 @@ def _get_dfs(self):
         if self.source_table is None:
             raise ValueError("source_table must be set")
 
-        input_df = read_csv(self.source_table, self.source_dtypes)
+        input_df = self._read_csv(self.source_table, dtype=self.source_dtypes)
         wdf = replace_nans(input_df)
         cleaned_df = filter_to_valid_point_ids(wdf)
         return input_df, cleaned_df