Formatting changes

Author: kbighorse

alembic/versions/95d8b982cd5d_add_nma_chemistry_lineage_relations.py

@@ -8,14 +8,15 @@
 1. thing_id FK column to NMA_Chemistry_SampleInfo (NOT NULL, no orphans)
 2. NMA_MinorTraceChemistry table (child of NMA_Chemistry_SampleInfo)
 """
+
 from typing import Sequence, Union
 
 from alembic import op
 import sqlalchemy as sa
 
 
-revision: str = '95d8b982cd5d'
-down_revision: Union[str, Sequence[str], None] = '2f6e9d3a1c45'
+revision: str = "95d8b982cd5d"
+down_revision: Union[str, Sequence[str], None] = "2f6e9d3a1c45"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
@@ -24,49 +25,46 @@ def upgrade() -> None:
     """Upgrade schema."""
     # Add thing_id FK to NMA_Chemistry_SampleInfo (NOT NULL - no orphans)
     op.add_column(
-        'NMA_Chemistry_SampleInfo',
-        sa.Column('thing_id', sa.Integer(), nullable=False)
+        "NMA_Chemistry_SampleInfo", sa.Column("thing_id", sa.Integer(), nullable=False)
     )
     op.create_foreign_key(
-        'fk_chemistry_sample_info_thing',
-        'NMA_Chemistry_SampleInfo',
-        'thing',
-        ['thing_id'],
-        ['id'],
-        ondelete='CASCADE'
+        "fk_chemistry_sample_info_thing",
+        "NMA_Chemistry_SampleInfo",
+        "thing",
+        ["thing_id"],
+        ["id"],
+        ondelete="CASCADE",
     )
 
     # Create NMA_MinorTraceChemistry table
     op.create_table(
-        'NMA_MinorTraceChemistry',
-        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
-        sa.Column('chemistry_sample_info_id', sa.Integer(), nullable=False),
-        sa.Column('analyte', sa.String(50), nullable=True),
-        sa.Column('sample_value', sa.Float(), nullable=True),
-        sa.Column('units', sa.String(20), nullable=True),
-        sa.Column('symbol', sa.String(10), nullable=True),
-        sa.Column('analysis_method', sa.String(100), nullable=True),
-        sa.Column('analysis_date', sa.Date(), nullable=True),
-        sa.Column('notes', sa.Text(), nullable=True),
-        sa.Column('analyses_agency', sa.String(100), nullable=True),
-        sa.Column('uncertainty', sa.Float(), nullable=True),
-        sa.Column('volume', sa.Float(), nullable=True),
-        sa.Column('volume_unit', sa.String(20), nullable=True),
-        sa.PrimaryKeyConstraint('id'),
+        "NMA_MinorTraceChemistry",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("chemistry_sample_info_id", sa.Integer(), nullable=False),
+        sa.Column("analyte", sa.String(50), nullable=True),
+        sa.Column("sample_value", sa.Float(), nullable=True),
+        sa.Column("units", sa.String(20), nullable=True),
+        sa.Column("symbol", sa.String(10), nullable=True),
+        sa.Column("analysis_method", sa.String(100), nullable=True),
+        sa.Column("analysis_date", sa.Date(), nullable=True),
+        sa.Column("notes", sa.Text(), nullable=True),
+        sa.Column("analyses_agency", sa.String(100), nullable=True),
+        sa.Column("uncertainty", sa.Float(), nullable=True),
+        sa.Column("volume", sa.Float(), nullable=True),
+        sa.Column("volume_unit", sa.String(20), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
         sa.ForeignKeyConstraint(
-            ['chemistry_sample_info_id'],
-            ['NMA_Chemistry_SampleInfo.OBJECTID'],
-            ondelete='CASCADE',
+            ["chemistry_sample_info_id"],
+            ["NMA_Chemistry_SampleInfo.OBJECTID"],
+            ondelete="CASCADE",
         ),
     )
 
 
 def downgrade() -> None:
     """Downgrade schema."""
-    op.drop_table('NMA_MinorTraceChemistry')
+    op.drop_table("NMA_MinorTraceChemistry")
     op.drop_constraint(
-        'fk_chemistry_sample_info_thing',
-        'NMA_Chemistry_SampleInfo',
-        type_='foreignkey'
+        "fk_chemistry_sample_info_thing", "NMA_Chemistry_SampleInfo", type_="foreignkey"
     )
-    op.drop_column('NMA_Chemistry_SampleInfo', 'thing_id')
+    op.drop_column("NMA_Chemistry_SampleInfo", "thing_id")

db/nma_legacy.py

@@ -218,7 +218,9 @@ class ChemistrySampleInfo(Base):
     def validate_thing_id(self, key, value):
         """Prevent orphan ChemistrySampleInfo - must have a parent Thing."""
         if value is None:
-            raise ValueError("ChemistrySampleInfo requires a parent Thing (thing_id cannot be None)")
+            raise ValueError(
+                "ChemistrySampleInfo requires a parent Thing (thing_id cannot be None)"
+            )
         return value
 
 
@@ -312,7 +314,9 @@ class NMAMinorTraceChemistry(Base):
     def validate_chemistry_sample_info_id(self, key, value):
         """Prevent orphan NMAMinorTraceChemistry - must have a parent ChemistrySampleInfo."""
         if value is None:
-            raise ValueError("NMAMinorTraceChemistry requires a parent ChemistrySampleInfo")
+            raise ValueError(
+                "NMAMinorTraceChemistry requires a parent ChemistrySampleInfo"
+            )
         return value
 
 

tests/test_nma_chemistry_lineage.py

@@ -463,7 +463,11 @@ def test_reverse_lineage_navigation(shared_well):
         session.refresh(well)
 
         # Reverse navigation - filter to just this sample_info
-        matching = [si for si in well.chemistry_sample_infos if si.object_id == sample_info.object_id]
+        matching = [
+            si
+            for si in well.chemistry_sample_infos
+            if si.object_id == sample_info.object_id
+        ]
         assert len(matching) == 1
         assert len(matching[0].minor_trace_chemistries) == 1
         assert matching[0].minor_trace_chemistries[0] == mtc

transfers/backfill/chemistry_sampleinfo.py

@@ -44,7 +44,7 @@ def __init__(self, *args, batch_size: int = 1000, **kwargs):
         # Cache Thing lookups to prevent N+1 queries
         self._thing_id_cache = {}
         self._build_thing_id_cache()
-    
+
     def _build_thing_id_cache(self):
         """Build cache of Thing.name -> thing.id to prevent orphan records."""
         with session_ctx() as session:
@@ -57,29 +57,29 @@ def _get_dfs(self) -> tuple[pd.DataFrame, pd.DataFrame]:
         # Filter to only include rows where Thing exists (prevent orphan records)
         cleaned_df = self._filter_to_valid_things(input_df)
         return input_df, cleaned_df
-    
+
     def _filter_to_valid_things(self, df: pd.DataFrame) -> pd.DataFrame:
         """
         Filter to only include rows where SamplePointID matches an existing Thing.
         Prevents orphan ChemistrySampleInfo records.
-        
+
         Uses cached Thing lookups for performance.
         """
         # Use cached Thing names (keys of thing_id_cache)
         valid_point_ids = set(self._thing_id_cache.keys())
-        
+
         # Filter to rows where SamplePointID exists as a Thing.name
         before_count = len(df)
         filtered_df = df[df["SamplePointID"].isin(valid_point_ids)].copy()
         after_count = len(filtered_df)
-        
+
         if before_count > after_count:
             skipped = before_count - after_count
             logger.warning(
                 f"Filtered out {skipped} ChemistrySampleInfo records without matching Things "
                 f"({after_count} valid, {skipped} orphan records prevented)"
             )
-        
+
         return filtered_df
 
     def _transfer_hook(self, session: Session) -> None:
@@ -97,13 +97,13 @@ def _transfer_hook(self, session: Session) -> None:
                 )
                 continue
             row_dicts.append(row_dict)
-        
+
         if skipped_count > 0:
             logger.warning(
                 f"Skipped {skipped_count} ChemistrySampleInfo records without valid Thing "
                 f"(orphan prevention)"
             )
-        
+
         rows = self._dedupe_rows(row_dicts, key="OBJECTID")
 
         insert_stmt = insert(ChemistrySampleInfo)