diff --git a/admin/config.py b/admin/config.py index 123495258..71d4499b9 100644 --- a/admin/config.py +++ b/admin/config.py @@ -38,6 +38,7 @@ SampleAdmin, HydraulicsDataAdmin, ChemistrySampleInfoAdmin, + FieldParametersAdmin, RadionuclidesAdmin, GeologicFormationAdmin, DataProvenanceAdmin, @@ -68,6 +69,7 @@ from db.nma_legacy import ( ChemistrySampleInfo, NMAHydraulicsData, + NMAFieldParameters, NMARadionuclides, SurfaceWaterData, ) @@ -138,6 +140,7 @@ def create_admin(app): # Samples admin.add_view(SampleAdmin(Sample)) admin.add_view(ChemistrySampleInfoAdmin(ChemistrySampleInfo)) + admin.add_view(FieldParametersAdmin(NMAFieldParameters)) admin.add_view(SurfaceWaterDataAdmin(SurfaceWaterData)) # Hydraulics diff --git a/admin/views/__init__.py b/admin/views/__init__.py index c5f0ec70e..3c93e783f 100644 --- a/admin/views/__init__.py +++ b/admin/views/__init__.py @@ -33,6 +33,7 @@ from admin.views.sample import SampleAdmin from admin.views.hydraulicsdata import HydraulicsDataAdmin from admin.views.chemistry_sampleinfo import ChemistrySampleInfoAdmin +from admin.views.field_parameters import FieldParametersAdmin from admin.views.radionuclides import RadionuclidesAdmin from admin.views.geologic_formation import GeologicFormationAdmin from admin.views.data_provenance import DataProvenanceAdmin @@ -62,6 +63,7 @@ "SampleAdmin", "HydraulicsDataAdmin", "ChemistrySampleInfoAdmin", + "FieldParametersAdmin", "RadionuclidesAdmin", "GeologicFormationAdmin", "DataProvenanceAdmin", diff --git a/admin/views/field_parameters.py b/admin/views/field_parameters.py new file mode 100644 index 000000000..a19a47ee0 --- /dev/null +++ b/admin/views/field_parameters.py @@ -0,0 +1,109 @@ +# =============================================================================== +# Copyright 2026 +# +# 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. +# =============================================================================== +""" +FieldParametersAdmin view for legacy NMA_FieldParameters. +""" +from admin.views.base import OcotilloModelView + + +class FieldParametersAdmin(OcotilloModelView): + """ + Admin view for NMAFieldParameters model. + """ + + # ========== Basic Configuration ========== + + name = "Field Parameters" + label = "Field Parameters" + icon = "fa fa-tachometer" + + can_create = False + can_edit = False + can_delete = False + + # ========== List View ========== + + list_fields = [ + "global_id", + "sample_pt_id", + "sample_point_id", + "field_parameter", + "sample_value", + "units", + "analyses_agency", + "wc_lab_id", + "object_id", + ] + + sortable_fields = [ + "global_id", + "sample_pt_id", + "sample_point_id", + "field_parameter", + "sample_value", + "units", + "notes", + "analyses_agency", + "wc_lab_id", + "object_id", + ] + + fields_default_sort = [("sample_point_id", True)] + + searchable_fields = [ + "global_id", + "sample_pt_id", + "sample_point_id", + "field_parameter", + "units", + "notes", + "analyses_agency", + "wc_lab_id", + ] + + page_size = 50 + page_size_options = [25, 50, 100, 200] + + # ========== Form View ========== + + fields = [ + "global_id", + "sample_pt_id", + "sample_point_id", + "field_parameter", + "sample_value", + "units", + "notes", + "object_id", + "analyses_agency", + "wc_lab_id", + ] + + field_labels = { + "global_id": "GlobalID", + "sample_pt_id": "SamplePtID", + "sample_point_id": "SamplePointID", + "field_parameter": "FieldParameter", + "sample_value": "SampleValue", + "units": "Units", + "notes": "Notes", + "object_id": "OBJECTID", + "analyses_agency": "AnalysesAgency", + "wc_lab_id": "WCLab_ID", + } + + +# ============= EOF ============================================= diff --git a/alembic/versions/c1d2e3f4a5b6_create_nma_field_parameters.py b/alembic/versions/c1d2e3f4a5b6_create_nma_field_parameters.py new file mode 100644 index 000000000..3708db371 --- /dev/null +++ b/alembic/versions/c1d2e3f4a5b6_create_nma_field_parameters.py @@ -0,0 +1,110 @@ +"""Create legacy NMA_FieldParameters table. + +Revision ID: c1d2e3f4a5b6 +Revises: 1d2c3b4a5e67 +Create Date: 2026-03-01 03:00:00.000000 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy import inspect +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = "c1d2e3f4a5b6" +down_revision: Union[str, Sequence[str], None] = "1d2c3b4a5e67" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Create the legacy field parameters table.""" + bind = op.get_bind() + inspector = inspect(bind) + if not inspector.has_table("NMA_FieldParameters"): + op.create_table( + "NMA_FieldParameters", + sa.Column( + "GlobalID", + postgresql.UUID(as_uuid=True), + nullable=False, + primary_key=True, + ), + sa.Column( + "SamplePtID", + postgresql.UUID(as_uuid=True), + sa.ForeignKey( + "NMA_Chemistry_SampleInfo.SamplePtID", + onupdate="CASCADE", + ondelete="CASCADE", + ), + nullable=False, + ), + sa.Column("SamplePointID", sa.String(length=10), nullable=True), + sa.Column("FieldParameter", sa.String(length=50), nullable=True), + sa.Column( + "SampleValue", sa.Float(), nullable=False, server_default=sa.text("0") + ), + sa.Column("Units", sa.String(length=50), nullable=True), + sa.Column("Notes", sa.String(length=255), nullable=True), + sa.Column( + "OBJECTID", + sa.Integer(), + sa.Identity(start=1), + nullable=False, + ), + sa.Column("AnalysesAgency", sa.String(length=50), nullable=True), + sa.Column("WCLab_ID", sa.String(length=25), nullable=True), + ) + op.create_index( + "FieldParameters$AnalysesAgency", + "NMA_FieldParameters", + ["AnalysesAgency"], + ) + op.create_index( + "FieldParameters$ChemistrySampleInfoFieldParameters", + "NMA_FieldParameters", + ["SamplePtID"], + ) + op.create_index( + "FieldParameters$FieldParameter", + "NMA_FieldParameters", + ["FieldParameter"], + ) + op.create_index( + "FieldParameters$SamplePointID", + "NMA_FieldParameters", + ["SamplePointID"], + ) + op.create_index( + "FieldParameters$SamplePtID", + "NMA_FieldParameters", + ["SamplePtID"], + ) + op.create_index( + "FieldParameters$WCLab_ID", + "NMA_FieldParameters", + ["WCLab_ID"], + ) + op.create_index( + "FieldParameters$GlobalID", + "NMA_FieldParameters", + ["GlobalID"], + unique=True, + ) + op.create_index( + "FieldParameters$OBJECTID", + "NMA_FieldParameters", + ["OBJECTID"], + unique=True, + ) + + +def downgrade() -> None: + """Drop the legacy field parameters table.""" + bind = op.get_bind() + inspector = inspect(bind) + if inspector.has_table("NMA_FieldParameters"): + op.drop_table("NMA_FieldParameters") diff --git a/db/nma_legacy.py b/db/nma_legacy.py index b76fad1e4..80890a1b5 100644 --- a/db/nma_legacy.py +++ b/db/nma_legacy.py @@ -33,6 +33,8 @@ Text, UniqueConstraint, text, + Identity, + Index, ) from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.orm import Mapped, mapped_column, relationship, validates @@ -314,6 +316,13 @@ class ChemistrySampleInfo(Base): passive_deletes=True, ) + field_parameters: Mapped[List["NMAFieldParameters"]] = relationship( + "NMAFieldParameters", + back_populates="chemistry_sample_info", + cascade="all, delete-orphan", + passive_deletes=True, + ) + @validates("thing_id") def validate_thing_id(self, key, value): """Prevent orphan ChemistrySampleInfo - must have a parent Thing.""" @@ -539,4 +548,75 @@ def validate_sample_pt_id(self, key, value): return value +class NMAFieldParameters(Base): + """ + Legacy FieldParameters table from AMPAPI. + Stores field measurements (pH, Temp, etc.) linked to ChemistrySampleInfo. + """ + + __tablename__ = "NMA_FieldParameters" + + __table_args__ = ( + # Explicit Indexes from DDL + Index("FieldParameters$AnalysesAgency", "AnalysesAgency"), + Index("FieldParameters$ChemistrySampleInfoFieldParameters", "SamplePtID"), + Index("FieldParameters$FieldParameter", "FieldParameter"), + Index("FieldParameters$SamplePointID", "SamplePointID"), + Index( + "FieldParameters$SamplePtID", "SamplePtID" + ), # Note: DDL had two indexes on this col + Index("FieldParameters$WCLab_ID", "WCLab_ID"), + # Unique Indexes (Explicitly named to match DDL) + Index("FieldParameters$GlobalID", "GlobalID", unique=True), + Index("FieldParameters$OBJECTID", "OBJECTID", unique=True), + ) + + # Primary Key + global_id: Mapped[uuid.UUID] = mapped_column( + "GlobalID", UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 + ) + + # Foreign Key + sample_pt_id: Mapped[uuid.UUID] = mapped_column( + "SamplePtID", + UUID(as_uuid=True), + ForeignKey( + "NMA_Chemistry_SampleInfo.SamplePtID", + onupdate="CASCADE", + ondelete="CASCADE", + ), + nullable=False, + ) + + # Legacy Columns + sample_point_id: Mapped[Optional[str]] = mapped_column("SamplePointID", String(10)) + field_parameter: Mapped[Optional[str]] = mapped_column("FieldParameter", String(50)) + sample_value: Mapped[float] = mapped_column( + "SampleValue", Float, server_default="0" + ) + units: Mapped[Optional[str]] = mapped_column("Units", String(50)) + notes: Mapped[Optional[str]] = mapped_column("Notes", String(255)) + + # Identity Column + object_id: Mapped[int] = mapped_column( + "OBJECTID", Integer, Identity(start=1), nullable=False + ) + + analyses_agency: Mapped[Optional[str]] = mapped_column("AnalysesAgency", String(50)) + wc_lab_id: Mapped[Optional[str]] = mapped_column("WCLab_ID", String(25)) + + # Relationships + chemistry_sample_info: Mapped["ChemistrySampleInfo"] = relationship( + "ChemistrySampleInfo", back_populates="field_parameters" + ) + + @validates("sample_pt_id") + def validate_sample_pt_id(self, key, value): + if value is None: + raise ValueError( + "FieldParameter requires a parent ChemistrySampleInfo (SamplePtID)" + ) + return value + + # ============= EOF ============================================= diff --git a/tests/test_field_parameters_legacy.py b/tests/test_field_parameters_legacy.py new file mode 100644 index 000000000..bc4b85402 --- /dev/null +++ b/tests/test_field_parameters_legacy.py @@ -0,0 +1,356 @@ +""" +Unit tests for FieldParameters legacy model. + +These tests verify the migration of columns from the legacy FieldParameters table. +Migrated columns (excluding SSMA_TimeStamp): +- SamplePtID -> sample_pt_id +- SamplePointID -> sample_point_id +- FieldParameter -> field_parameter +- SampleValue -> sample_value +- Units -> units +- Notes -> notes +- OBJECTID -> object_id +- GlobalID -> global_id +- AnalysesAgency -> analyses_agency +- WCLab_ID -> wc_lab_id +""" + +import pytest +from uuid import uuid4 +from sqlalchemy import select, inspect +from sqlalchemy.exc import IntegrityError, ProgrammingError + +from db.engine import session_ctx +from db.nma_legacy import ChemistrySampleInfo, NMAFieldParameters + + +def _next_sample_point_id() -> str: + return f"SP-{uuid4().hex[:7]}" + + +def _create_sample_info(session, water_well_thing) -> ChemistrySampleInfo: + sample = ChemistrySampleInfo( + sample_pt_id=uuid4(), + sample_point_id=_next_sample_point_id(), + thing_id=water_well_thing.id, + ) + session.add(sample) + session.commit() + return sample + + +# ===================== Table and Column Existence Tests ========================== + + +def test_field_parameters_has_all_migrated_columns(): + """ + VERIFIES: The SQLAlchemy model matches the migration mapping contract. + This ensures all Python-side attribute names exist as expected in the ORM. + """ + mapper = inspect(NMAFieldParameters) + actual_columns = [column.key for column in mapper.attrs] + + expected_columns = [ + "global_id", + "sample_pt_id", + "sample_point_id", + "field_parameter", + "sample_value", + "units", + "notes", + "object_id", + "analyses_agency", + "wc_lab_id", + ] + + for column in expected_columns: + assert column in actual_columns, f"Model is missing expected column: {column}" + + +def test_field_parameters_table_name(): + """Test that the table name follows convention.""" + assert NMAFieldParameters.__tablename__ == "NMA_FieldParameters" + + +# ===================== Functional & CRUD Tests ========================= + + +def test_field_parameters_persistence(water_well_thing): + """ + Verifies that data correctly persists and retrieves for the core columns. + This confirms the Postgres data types (REAL, UUID, VARCHAR) are compatible. + """ + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + test_global_id = uuid4() + new_fp = NMAFieldParameters( + global_id=test_global_id, + sample_pt_id=sample_info.sample_pt_id, + sample_point_id="PT-123", + field_parameter="pH", + sample_value=7.4, + units="SU", + notes="Legacy migration verification", + analyses_agency="NMA Agency", + wc_lab_id="WCLAB-01", + ) + + session.add(new_fp) + session.commit() + session.expire_all() + + retrieved = session.get(NMAFieldParameters, test_global_id) + assert retrieved.sample_value == 7.4 + assert retrieved.field_parameter == "pH" + assert retrieved.units == "SU" + assert retrieved.analyses_agency == "NMA Agency" + + session.delete(new_fp) + session.delete(sample_info) + session.commit() + + +def test_object_id_auto_generation(water_well_thing): + """Verifies that the OBJECTID (Identity) column auto-increments in Postgres.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + fp1 = NMAFieldParameters( + sample_pt_id=sample_info.sample_pt_id, + field_parameter="Temp", + ) + session.add(fp1) + session.commit() + session.refresh(fp1) + + assert fp1.object_id is not None + + session.delete(fp1) + session.delete(sample_info) + session.commit() + + +# ===================== CREATE tests ========================== +def test_create_field_parameters_all_fields(water_well_thing): + """Test creating a field parameters record with all fields.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + record = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + sample_point_id=sample_info.sample_point_id, + field_parameter="pH", + sample_value=7.4, + units="SU", + notes="Test notes", + analyses_agency="NMBGMR", + wc_lab_id="LAB-202", + ) + session.add(record) + session.commit() + session.refresh(record) + + assert record.global_id is not None + assert record.sample_pt_id == sample_info.sample_pt_id + assert record.sample_point_id == sample_info.sample_point_id + assert record.field_parameter == "pH" + assert record.sample_value == 7.4 + + session.delete(record) + session.delete(sample_info) + session.commit() + + +def test_create_field_parameters_minimal(water_well_thing): + """Test creating a field parameters record with minimal fields.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + record = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + ) + session.add(record) + session.commit() + session.refresh(record) + + assert record.global_id is not None + assert record.sample_pt_id == sample_info.sample_pt_id + assert record.field_parameter is None + assert record.units is None + assert record.sample_value == 0 + + session.delete(record) + session.delete(sample_info) + session.commit() + + +# ===================== READ tests ========================== +def test_read_field_parameters_by_global_id(water_well_thing): + """Test reading a field parameters record by GlobalID.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + record = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + ) + session.add(record) + session.commit() + + fetched = session.get(NMAFieldParameters, record.global_id) + assert fetched is not None + assert fetched.global_id == record.global_id + + session.delete(record) + session.delete(sample_info) + session.commit() + + +def test_query_field_parameters_by_sample_point_id(water_well_thing): + """Test querying field parameters by sample_point_id.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + record1 = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + sample_point_id=sample_info.sample_point_id, + ) + record2 = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + sample_point_id="OTHER-PT", + ) + session.add_all([record1, record2]) + session.commit() + + # Use SQLAlchemy 2.0 style select/execute for ORM queries. + stmt = select(NMAFieldParameters).filter( + NMAFieldParameters.sample_point_id == sample_info.sample_point_id + ) + results = session.execute(stmt).scalars().all() + assert len(results) >= 1 + assert all(r.sample_point_id == sample_info.sample_point_id for r in results) + + session.delete(record1) + session.delete(record2) + session.delete(sample_info) + session.commit() + + +# ===================== UPDATE tests ========================== +def test_update_field_parameters(water_well_thing): + """Test updating a field parameters record.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + record = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + ) + session.add(record) + session.commit() + + record.analyses_agency = "Updated Agency" + record.notes = "Updated notes" + session.commit() + session.refresh(record) + + assert record.analyses_agency == "Updated Agency" + assert record.notes == "Updated notes" + + session.delete(record) + session.delete(sample_info) + session.commit() + + +# ===================== DELETE tests ========================== +def test_delete_field_parameters(water_well_thing): + """Test deleting a field parameters record.""" + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + record = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + ) + session.add(record) + session.commit() + + session.delete(record) + session.commit() + + fetched = session.get(NMAFieldParameters, record.global_id) + assert fetched is None + + session.delete(sample_info) + session.commit() + + +# ===================== Relational Integrity Tests ====================== + + +def test_orphan_prevention_constraint(): + """ + VERIFIES: 'SamplePtID IS NOT NULL' and Foreign Key presence. + Ensures the DB rejects records that aren't linked to a ChemistrySampleInfo. + """ + with session_ctx() as session: + orphan = NMAFieldParameters( + field_parameter="pH", + sample_value=7.0, + ) + session.add(orphan) + + with pytest.raises((IntegrityError, ProgrammingError)): + session.flush() + session.rollback() + + +def test_cascade_delete_behavior(water_well_thing): + """ + VERIFIES: 'on delete cascade' behavior. + Deleting the parent sample must automatically remove associated field measurements. + """ + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + fp = NMAFieldParameters( + sample_pt_id=sample_info.sample_pt_id, + field_parameter="Temperature", + ) + session.add(fp) + session.commit() + session.refresh(fp) + fp_id = fp.global_id + + # Delete parent and check child + session.delete(sample_info) + session.commit() + session.expire_all() + + assert ( + session.get(NMAFieldParameters, fp_id) is None + ), "Child record persisted after parent deletion." + + +def test_update_cascade_propagation(water_well_thing): + """ + VERIFIES: foreign key integrity on SamplePtID. + Ensures the DB rejects updates to a non-existent parent SamplePtID. + """ + with session_ctx() as session: + sample_info = _create_sample_info(session, water_well_thing) + fp = NMAFieldParameters( + global_id=uuid4(), + sample_pt_id=sample_info.sample_pt_id, + field_parameter="Dissolved Oxygen", + ) + session.add(fp) + session.commit() + fp_id = fp.global_id + + with pytest.raises((IntegrityError, ProgrammingError)): + fp.sample_pt_id = uuid4() + session.flush() + session.rollback() + + fetched = session.get(NMAFieldParameters, fp_id) + if fetched is not None: + session.delete(fetched) + session.delete(sample_info) + session.commit() diff --git a/transfers/field_parameters_transfer.py b/transfers/field_parameters_transfer.py new file mode 100644 index 000000000..885e9fdab --- /dev/null +++ b/transfers/field_parameters_transfer.py @@ -0,0 +1,241 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Transfer FieldParameters data from NM_Aquifer to NMA_FieldParameters. + +This transfer requires ChemistrySampleInfo to be backfilled first. Each +FieldParameters record links to a ChemistrySampleInfo record via SamplePtID. +""" + +from __future__ import annotations + +from typing import Any, Optional +from uuid import UUID + +import pandas as pd +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.orm import Session + +from db import ChemistrySampleInfo, NMAFieldParameters +from db.engine import session_ctx +from transfers.logger import logger +from transfers.transferer import Transferer +from transfers.util import read_csv + + +class FieldParametersTransferer(Transferer): + """ + Transfer FieldParameters records to NMA_FieldParameters. + + Looks up ChemistrySampleInfo by SamplePtID and creates linked + NMAFieldParameters records. Uses upsert for idempotent transfers. + """ + + source_table = "FieldParameters" + + def __init__(self, *args, batch_size: int = 1000, **kwargs): + super().__init__(*args, **kwargs) + self.batch_size = batch_size + self._sample_pt_ids: set[UUID] = set() + self._build_sample_pt_id_cache() + + def _build_sample_pt_id_cache(self) -> None: + """Build cache of ChemistrySampleInfo.SamplePtID values.""" + with session_ctx() as session: + sample_infos = session.query(ChemistrySampleInfo.sample_pt_id).all() + self._sample_pt_ids = {sample_pt_id for (sample_pt_id,) in sample_infos} + logger.info( + f"Built ChemistrySampleInfo cache with {len(self._sample_pt_ids)} entries" + ) + + def _get_dfs(self) -> tuple[pd.DataFrame, pd.DataFrame]: + input_df = read_csv(self.source_table) + cleaned_df = self._filter_to_valid_sample_infos(input_df) + return input_df, cleaned_df + + def _filter_to_valid_sample_infos(self, df: pd.DataFrame) -> pd.DataFrame: + """ + Filter to only include rows where SamplePtID matches a ChemistrySampleInfo. + + This prevents orphan records and ensures the FK constraint will be satisfied. + """ + valid_sample_pt_ids = self._sample_pt_ids + before_count = len(df) + mask = df["SamplePtID"].apply( + lambda value: self._uuid_val(value) in valid_sample_pt_ids + ) + filtered_df = df[mask].copy() + after_count = len(filtered_df) + + if before_count > after_count: + skipped = before_count - after_count + logger.warning( + f"Filtered out {skipped} FieldParameters records without matching " + f"ChemistrySampleInfo ({after_count} valid, {skipped} orphan records prevented)" + ) + + return filtered_df + + def _transfer_hook(self, session: Session) -> None: + """ + Override transfer hook to use batch upsert for idempotent transfers. + + Uses ON CONFLICT DO UPDATE on GlobalID. + """ + limit = self.flags.get("LIMIT", 0) + df = self.cleaned_df + if limit > 0: + df = df.head(limit) + + row_dicts = [] + for row in df.itertuples(): + row_dict = self._row_to_dict(row) + if row_dict is not None: + row_dicts.append(row_dict) + + if not row_dicts: + logger.warning("No valid rows to transfer") + return + + rows = self._dedupe_rows(row_dicts) + logger.info(f"Upserting {len(rows)} FieldParameters records") + + insert_stmt = insert(NMAFieldParameters) + excluded = insert_stmt.excluded + + for i in range(0, len(rows), self.batch_size): + chunk = rows[i : i + self.batch_size] + logger.info(f"Upserting batch {i}-{i+len(chunk)-1} ({len(chunk)} rows)") + stmt = insert_stmt.values(chunk).on_conflict_do_update( + index_elements=["GlobalID"], + set_={ + "SamplePtID": excluded.SamplePtID, + "SamplePointID": excluded.SamplePointID, + "FieldParameter": excluded.FieldParameter, + "SampleValue": excluded.SampleValue, + "Units": excluded.Units, + "Notes": excluded.Notes, + "OBJECTID": excluded.OBJECTID, + "AnalysesAgency": excluded.AnalysesAgency, + "WCLab_ID": excluded.WCLab_ID, + }, + ) + session.execute(stmt) + session.commit() + session.expunge_all() + + def _row_to_dict(self, row) -> Optional[dict[str, Any]]: + """Convert a DataFrame row to a dict for upsert.""" + sample_pt_id = self._uuid_val(getattr(row, "SamplePtID", None)) + if sample_pt_id is None: + self._capture_error( + getattr(row, "SamplePtID", None), + f"Invalid SamplePtID: {getattr(row, 'SamplePtID', None)}", + "SamplePtID", + ) + return None + + if sample_pt_id not in self._sample_pt_ids: + self._capture_error( + sample_pt_id, + f"ChemistrySampleInfo not found for SamplePtID: {sample_pt_id}", + "SamplePtID", + ) + return None + + global_id = self._uuid_val(getattr(row, "GlobalID", None)) + if global_id is None: + self._capture_error( + getattr(row, "GlobalID", None), + f"Invalid GlobalID: {getattr(row, 'GlobalID', None)}", + "GlobalID", + ) + return None + + return { + "GlobalID": global_id, + "SamplePtID": sample_pt_id, + "SamplePointID": self._safe_str(row, "SamplePointID"), + "FieldParameter": self._safe_str(row, "FieldParameter"), + "SampleValue": self._safe_float(row, "SampleValue"), + "Units": self._safe_str(row, "Units"), + "Notes": self._safe_str(row, "Notes"), + "OBJECTID": self._safe_int(row, "OBJECTID"), + "AnalysesAgency": self._safe_str(row, "AnalysesAgency"), + "WCLab_ID": self._safe_str(row, "WCLab_ID"), + } + + def _dedupe_rows(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Dedupe rows by unique key to avoid ON CONFLICT loops. Later rows win.""" + deduped = {} + for row in rows: + key = row.get("GlobalID") + if key is None: + continue + deduped[key] = row + return list(deduped.values()) + + def _safe_str(self, row, attr: str) -> Optional[str]: + """Safely get a string value, returning None for NaN.""" + val = getattr(row, attr, None) + if val is None or pd.isna(val): + return None + return str(val) + + def _safe_float(self, row, attr: str) -> Optional[float]: + """Safely get a float value, returning None for NaN.""" + val = getattr(row, attr, None) + if val is None or pd.isna(val): + return None + try: + return float(val) + except (TypeError, ValueError): + return None + + def _safe_int(self, row, attr: str) -> Optional[int]: + """Safely get an int value, returning None for NaN.""" + val = getattr(row, attr, None) + if val is None or pd.isna(val): + return None + try: + return int(val) + except (TypeError, ValueError): + return None + + def _uuid_val(self, value: Any) -> Optional[UUID]: + if value is None or pd.isna(value): + return None + if isinstance(value, UUID): + return value + if isinstance(value, str): + try: + return UUID(value) + except ValueError: + return None + return None + + +def run(flags: dict = None) -> tuple[pd.DataFrame, pd.DataFrame, list]: + """Entrypoint to execute the transfer.""" + transferer = FieldParametersTransferer(flags=flags) + transferer.transfer() + return transferer.input_df, transferer.cleaned_df, transferer.errors + + +if __name__ == "__main__": + # Allow running via `python -m transfers.field_parameters_transfer` + run() + +# ============= EOF =============================================