BDMS-470: Migrate Stratigraphy

alembic/versions/1d2c3b4a5e67_create_nma_stratigraphy_table.py

@@ -0,0 +1,70 @@
+"""Create legacy NMA_Stratigraphy table.
+
+Revision ID: 1d2c3b4a5e67
+Revises: a7b8c9d0e1f2
+Create Date: 2026-01-15 00: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 = "1d2c3b4a5e67"
+down_revision: Union[str, Sequence[str], None] = "a7b8c9d0e1f2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create the legacy stratigraphy table."""
+    bind = op.get_bind()
+    inspector = inspect(bind)
+    if inspector.has_table("NMA_Stratigraphy"):
+        return
+
+    op.create_table(
+        "NMA_Stratigraphy",
+        sa.Column(
+            "GlobalID",
+            postgresql.UUID(as_uuid=True),
+            primary_key=True,
+            nullable=False,
+        ),
+        sa.Column("WellID", postgresql.UUID(as_uuid=True), nullable=True),
+        sa.Column("PointID", sa.String(length=10), nullable=False),
+        sa.Column(
+            "thing_id",
+            sa.Integer(),
+            sa.ForeignKey("thing.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("StratTop", sa.Float(), nullable=True),
+        sa.Column("StratBottom", sa.Float(), nullable=True),
+        sa.Column("UnitIdentifier", sa.String(length=50), nullable=True),
+        sa.Column("Lithology", sa.String(length=100), nullable=True),
+        sa.Column("LithologicModifier", sa.String(length=100), nullable=True),
+        sa.Column("ContributingUnit", sa.String(length=10), nullable=True),
+        sa.Column("StratSource", sa.Text(), nullable=True),
+        sa.Column("StratNotes", sa.Text(), nullable=True),
+        sa.Column("OBJECTID", sa.Integer(), nullable=True, unique=True),
+    )
+    op.create_index(
+        "ix_nma_stratigraphy_point_id",
+        "NMA_Stratigraphy",
+        ["PointID"],
+    )
+    op.create_index(
+        "ix_nma_stratigraphy_thing_id",
+        "NMA_Stratigraphy",
+        ["thing_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_nma_stratigraphy_thing_id", table_name="NMA_Stratigraphy")
+    op.drop_index("ix_nma_stratigraphy_point_id", table_name="NMA_Stratigraphy")
+    op.drop_table("NMA_Stratigraphy")

db/nma_legacy.py

@@ -17,10 +17,10 @@
 """Legacy NM Aquifer models copied from AMPAPI."""
 
 import uuid
-
 from datetime import date, datetime
 from typing import TYPE_CHECKING, List, Optional
 
+from db.base import Base
 from sqlalchemy import (
     Boolean,
     Date,
@@ -37,8 +37,6 @@
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
 
-from db.base import Base
-
 if TYPE_CHECKING:
     from db.thing import Thing
 
@@ -206,6 +204,37 @@ class NMAHydraulicsData(Base):
     thing: Mapped["Thing"] = relationship("Thing")
 
 
+class Stratigraphy(Base):
+    """Legacy stratigraphy (lithology log) data from AMPAPI."""
+
+    __tablename__ = "NMA_Stratigraphy"
+
+    global_id: Mapped[uuid.UUID] = mapped_column(
+        "GlobalID", UUID(as_uuid=True), primary_key=True
+    )
+    well_id: Mapped[Optional[uuid.UUID]] = mapped_column("WellID", UUID(as_uuid=True))
+    point_id: Mapped[str] = mapped_column("PointID", String(10), nullable=False)
+    thing_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("thing.id", ondelete="CASCADE"), nullable=False
+    )
+
+    strat_top: Mapped[Optional[float]] = mapped_column("StratTop", Float)
+    strat_bottom: Mapped[Optional[float]] = mapped_column("StratBottom", Float)
+    unit_identifier: Mapped[Optional[str]] = mapped_column("UnitIdentifier", String(50))
+    lithology: Mapped[Optional[str]] = mapped_column("Lithology", String(100))
+    lithologic_modifier: Mapped[Optional[str]] = mapped_column(
+        "LithologicModifier", String(100)
+    )
+    contributing_unit: Mapped[Optional[str]] = mapped_column(
+        "ContributingUnit", String(10)
+    )
+    strat_source: Mapped[Optional[str]] = mapped_column("StratSource", Text)
+    strat_notes: Mapped[Optional[str]] = mapped_column("StratNotes", Text)
+    object_id: Mapped[Optional[int]] = mapped_column("OBJECTID", Integer, unique=True)
+
+    thing: Mapped["Thing"] = relationship("Thing", back_populates="stratigraphy_logs")
+
+
 class ChemistrySampleInfo(Base):
     """
     Legacy Chemistry SampleInfo table from AMPAPI.

db/thing.py

@@ -16,11 +16,6 @@
 from datetime import date
 from typing import List, TYPE_CHECKING
 
-from sqlalchemy import Integer, ForeignKey, String, Column, Float, Date
-from sqlalchemy.ext.associationproxy import association_proxy, AssociationProxy
-from sqlalchemy.orm import relationship, mapped_column, Mapped
-from sqlalchemy_utils import TSVectorType
-
 from db import lexicon_term, NotesMixin
 from db.asset import Asset
 from db.base import (
@@ -33,6 +28,10 @@
 from db.permission_history import PermissionHistoryMixin
 from db.status_history import StatusHistoryMixin
 from services.util import retrieve_latest_polymorphic_history_table_record
+from sqlalchemy import Integer, ForeignKey, String, Column, Float, Date
+from sqlalchemy.ext.associationproxy import association_proxy, AssociationProxy
+from sqlalchemy.orm import relationship, mapped_column, Mapped
+from sqlalchemy_utils import TSVectorType
 
 if TYPE_CHECKING:
     from db.location import Location
@@ -47,7 +46,7 @@
     from db.thing_geologic_formation_association import (
         ThingGeologicFormationAssociation,
     )
-    from db.nma_legacy import ChemistrySampleInfo
+    from db.nma_legacy import ChemistrySampleInfo, Stratigraphy
 
 
 class Thing(
@@ -312,6 +311,13 @@ class Thing(
         passive_deletes=True,
     )
 
+    stratigraphy_logs: Mapped[List["Stratigraphy"]] = relationship(
+        "Stratigraphy",
+        back_populates="thing",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+    )
+
     # --- Association Proxies ---
     assets: AssociationProxy[list["Asset"]] = association_proxy(
         "asset_associations", "asset"
@@ -417,6 +423,8 @@ def measuring_point_height(self) -> int | None:
         Since measuring_point_history is eagerly loaded, this should not introduce N+1 query issues.
         """
         if self.thing_type == "water well":
+            if not self.measuring_points:
+                return None
             sorted_measuring_point_history = sorted(
                 self.measuring_points, key=lambda x: x.start_date, reverse=True
             )
@@ -433,6 +441,8 @@ def measuring_point_description(self) -> str | None:
         Since measuring_point_history is eagerly loaded, this should not introduce N+1 query issues.
         """
         if self.thing_type == "water well":
+            if not self.measuring_points:
+                return None
             sorted_measuring_point_history = sorted(
                 self.measuring_points, key=lambda x: x.start_date, reverse=True
             )

schemas/location.py

@@ -17,12 +17,11 @@
 from typing import Any
 from typing import List
 
+from core.constants import SRID_WGS84, SRID_UTM_ZONE_13N
+from core.enums import ElevationMethod, CoordinateMethod
 from geoalchemy2 import WKBElement
 from geoalchemy2.shape import to_shape
 from pydantic import BaseModel, model_validator, field_validator, Field, ConfigDict
-
-from core.constants import SRID_WGS84, SRID_UTM_ZONE_13N
-from core.enums import ElevationMethod, CoordinateMethod
 from schemas import BaseCreateModel, BaseUpdateModel, BaseResponseModel
 from schemas.notes import NoteResponse, CreateNote, UpdateNote
 from services.util import convert_m_to_ft, transform_srid
@@ -130,6 +129,8 @@ class LocationGeoJSONResponse(BaseModel):
     @model_validator(mode="before")
     @classmethod
     def populate_fields(cls, data: Any) -> Any:
+        if data is None:
+            return None
         # convert row to dictionary
         if not isinstance(data, dict):
             data_dict = {c.name: getattr(data, c.name) for c in data.__table__.columns}

tests/conftest.py

@@ -1,23 +1,19 @@
 import os
-import uuid
 
 import pytest
 from alembic import command
 from alembic.config import Config
-from dotenv import load_dotenv
-from sqlalchemy import text
-from sqlalchemy_utils import TSVectorType
-from sqlalchemy_searchable import sync_trigger
-
 from core.initializers import init_lexicon, init_parameter
 from db import *
 from db.engine import session_ctx
+from dotenv import load_dotenv
+from sqlalchemy_searchable import sync_trigger
 from tests import get_parameter_id
 
 
 def pytest_configure():
     load_dotenv(override=True)
-    os.environ.setdefault("POSTGRES_PORT", "54321")
+    os.environ["POSTGRES_PORT"] = "54321"
 
 
 def _alembic_config() -> Config:
@@ -72,9 +68,54 @@ def _setup_test_db():
     _sync_search_vectors()
     init_lexicon()
     init_parameter()
+    _ensure_locations_for_things()
+    _ensure_measuring_points_for_wells()
     yield
 
 
+def _ensure_locations_for_things() -> None:
+    with session_ctx() as session:
+        things = session.query(Thing).all()
+        for thing in things:
+            if thing.location_associations:
+                continue
+            location = Location(
+                point="POINT(-107.0 33.0)",
+                elevation=0,
+                release_status=thing.release_status,
+            )
+            session.add(location)
+            session.flush()
+            session.add(
+                LocationThingAssociation(
+                    location_id=location.id,
+                    thing_id=thing.id,
+                    effective_start=date.today().isoformat(),
+                )
+            )
+        session.commit()
+
+
+def _ensure_measuring_points_for_wells() -> None:
+    with session_ctx() as session:
+        wells = session.query(Thing).filter(Thing.thing_type == "water well").all()
+        missing = [well for well in wells if not well.measuring_points]
+        for well in missing:
+            session.add(
+                MeasuringPointHistory(
+                    thing_id=well.id,
+                    measuring_point_height=well.well_casing_depth
+                    or well.well_depth
+                    or 0.0,
+                    measuring_point_description="autogenerated for tests",
+                    start_date=date.today(),
+                    release_status=well.release_status,
+                )
+            )
+        if missing:
+            session.commit()
+
+
 @pytest.fixture()
 def location():
     with session_ctx() as session: