Merge pull request #501 from DataIntegrationGroup/transfer-fix

transfer-fix

Author: jirhiker

transfers/minor_trace_chemistry_transfer.py

@@ -114,7 +114,8 @@ def _transfer_hook(self, session: Session) -> None:
         """
         Override transfer hook to use batch upsert for idempotent transfers.
 
-        Uses ON CONFLICT DO UPDATE on nma_GlobalID (the legacy UUID PK, now UNIQUE).
+        Uses ON CONFLICT DO UPDATE on (chemistry_sample_info_id, analyte),
+        matching uq_minor_trace_chemistry_sample_analyte.
         """
         df = self.cleaned_df
 
@@ -129,8 +130,12 @@ def _transfer_hook(self, session: Session) -> None:
             logger.warning("No valid rows to transfer")
             return
 
-        # Dedupe by nma_GlobalID to avoid PK conflicts.
-        rows = self._dedupe_rows(row_dicts)
+        # Dedupe by the same logical key used by the table unique constraint.
+        rows = self._dedupe_rows(
+            row_dicts,
+            key=["chemistry_sample_info_id", "analyte"],
+            include_missing=True,
+        )
         logger.info(f"Upserting {len(rows)} MinorTraceChemistry records")
 
         insert_stmt = insert(NMA_MinorTraceChemistry)
@@ -139,9 +144,9 @@ def _transfer_hook(self, session: Session) -> None:
         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)")
-            # Upsert on nma_GlobalID (legacy UUID PK, now UNIQUE)
+            # Upsert on unique logical key (chemistry_sample_info_id, analyte)
             stmt = insert_stmt.values(chunk).on_conflict_do_update(
-                index_elements=["nma_GlobalID"],
+                index_elements=["chemistry_sample_info_id", "analyte"],
                 set_={
                     "chemistry_sample_info_id": excluded.chemistry_sample_info_id,
                     "nma_chemistry_sample_info_uuid": excluded.nma_chemistry_sample_info_uuid,