Transfer

.gitignore

@@ -32,6 +32,7 @@ transfers/data/nma_csv_cache/*
 tests/features/*.feature
 transfers/metrics/*
 transfers/logs/*
+run_bdd-local.sh
 
 
 # deployment files

.pre-commit-config.yaml

@@ -16,15 +16,15 @@ repos:
           '--statistics'
           ]
         exclude: ^db/__init__.py$ # all models need to be imported for Alembic, but are not used directly
-#  - repo: local
-#    hooks:
-#      - id: pytest
-#        name: pytest
-#        entry: pytest  # Or your specific test command, e.g., poetry run pytest
-#        language: system
-#        types: [python] # Specify relevant file types for your tests
-#        pass_filenames: false
-#        always_run: true
+  - repo: local
+    hooks:
+      - id: pytest
+        name: pytest
+        entry: pytest  # Or your specific test command, e.g., poetry run pytest
+        language: system
+        types: [python] # Specify relevant file types for your tests
+        pass_filenames: false
+        always_run: true
 
   # - repo: https://github.com/pre-commit/mirrors-mypy
   #   rev: v1.10.0  # Use the latest stable version or pin to your preference

core/app.py

@@ -17,8 +17,6 @@
 from contextlib import asynccontextmanager
 from typing import AsyncGenerator
 
-from db.engine import session_ctx
-
 from fastapi import FastAPI
 from fastapi.openapi.docs import (
     get_swagger_ui_html,
@@ -27,8 +25,6 @@
 from fastapi.openapi.utils import get_openapi
 
 from .initializers import (
-    init_lexicon,
-    init_parameter,
     register_routes,
     erase_and_rebuild_db,
 )
@@ -41,10 +37,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     Application lifespan event handler to initialize the database and lexicon.
     """
     if settings.get_enum("MODE") == "development":
-        with session_ctx() as session:
-            erase_and_rebuild_db(session)
-        init_lexicon()
-        init_parameter()
+        erase_and_rebuild_db()
 
     register_routes(app)
     yield

core/initializers.py

@@ -18,7 +18,6 @@
 from fastapi_pagination import add_pagination
 from sqlalchemy import text
 from sqlalchemy.exc import DatabaseError
-from sqlalchemy.orm import Session
 
 from db import Base
 from db.engine import session_ctx
@@ -56,13 +55,14 @@ def init_parameter(path: str = None) -> None:
                 session.rollback()
 
 
-def erase_and_rebuild_db(session: Session):
-    session.execute(text("DROP SCHEMA public CASCADE"))
-    session.execute(text("CREATE SCHEMA public"))
-    session.execute(text("CREATE EXTENSION IF NOT EXISTS postgis"))
-    session.commit()
-    Base.metadata.drop_all(session.bind)
-    Base.metadata.create_all(session.bind)
+def erase_and_rebuild_db():
+    with session_ctx() as session:
+        session.execute(text("DROP SCHEMA public CASCADE"))
+        session.execute(text("CREATE SCHEMA public"))
+        session.execute(text("CREATE EXTENSION IF NOT EXISTS postgis"))
+        session.commit()
+        Base.metadata.drop_all(session.bind)
+        Base.metadata.create_all(session.bind)
 
     init_lexicon()
     init_parameter()

schemas/__init__.py

@@ -13,14 +13,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ===============================================================================
-from datetime import datetime, timezone
+from datetime import datetime, timezone, date
+from typing import Annotated
 
 from pydantic import (
     BaseModel,
     ConfigDict,
     AwareDatetime,
     field_validator,
 )
+from pydantic.functional_validators import AfterValidator
 from pydantic.json_schema import JsonSchemaValue
 from pydantic_core import core_schema
 
@@ -51,6 +53,15 @@ class BaseUpdateModel(BaseCreateModel):
     release_status: ReleaseStatus | None = None
 
 
+def past_or_today_validator(value: date) -> date:
+    if value > date.today():
+        raise ValueError("Date must be today or in the past.")
+    return value
+
+
+PastOrTodayDate = Annotated[date, AfterValidator(past_or_today_validator)]
+
+
 # Custom type for UTC datetime serialization
 class UTCAwareDatetime(AwareDatetime):
     """Custom datetime type that always serializes to UTC with 'Z' suffix."""

schemas/thing.py

@@ -15,7 +15,7 @@
 # ===============================================================================
 from typing import List
 
-from pydantic import BaseModel, model_validator, PastDate, Field, field_validator
+from pydantic import BaseModel, model_validator, Field, field_validator
 
 from core.enums import (
     WellPurpose,
@@ -25,9 +25,9 @@
     Organization,
     MonitoringFrequency,
 )
-from schemas import BaseCreateModel, BaseUpdateModel, BaseResponseModel
-from schemas.location import LocationGeoJSONResponse
+from schemas import BaseCreateModel, BaseUpdateModel, BaseResponseModel, PastOrTodayDate
 from schemas.group import GroupResponse
+from schemas.location import LocationGeoJSONResponse
 from schemas.notes import NoteResponse, CreateNote
 
 
@@ -102,7 +102,7 @@ class CreateBaseThing(BaseCreateModel):
     location_id: int | None
     group_id: int | None = None  # Optional group ID for the thing
     name: str  # Name of the thing
-    first_visit_date: PastDate | None = None  # Date of NMBGMR's first visit
+    first_visit_date: PastOrTodayDate | None = None  # Date of NMBGMR's first visit
 
 
 class CreateWell(CreateBaseThing, ValidateWell):
@@ -171,15 +171,15 @@ class ThingIdLinkResponse(BaseResponseModel):
 
 class MonitoringFrequencyResponse(BaseModel):
     monitoring_frequency: MonitoringFrequency
-    start_date: PastDate
-    end_date: PastDate | None
+    start_date: PastOrTodayDate
+    end_date: PastOrTodayDate | None
 
 
 class BaseThingResponse(BaseResponseModel):
     name: str
     thing_type: str
     current_location: LocationGeoJSONResponse
-    first_visit_date: PastDate | None
+    first_visit_date: PastOrTodayDate | None
     # The new relationship to the polymorphic Notes table
     notes: List[NoteResponse] = []
 
@@ -317,7 +317,7 @@ class UpdateThing(BaseUpdateModel):
     """
 
     name: str | None = None  # Optional name for the thing
-    first_visit_date: PastDate | None = None  # Date of NMBGMR's first visit
+    first_visit_date: PastOrTodayDate | None = None  # Date of NMBGMR's first visit
 
 
 class UpdateWell(UpdateThing, ValidateWell):

tests/__init__.py

@@ -41,11 +41,9 @@
 from db.engine import session_ctx
 from core.app import app
 
-with session_ctx() as session:
-    erase_and_rebuild_db(session)
-
-
+erase_and_rebuild_db()
 register_routes(app)
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],  # Allows all origins, adjust as needed for security

tests/features/environment.py

@@ -357,9 +357,10 @@ def before_all(context):
     context.objects = {}
     rebuild = False
     # rebuild = True
+    if rebuild:
+        erase_and_rebuild_db()
+
     with session_ctx() as session:
-        if rebuild:
-            erase_and_rebuild_db(session)
 
         loc_1 = add_location(context, session)
         loc_2 = add_location(context, session)

transfers/contact_transfer.py

@@ -61,7 +61,8 @@ def transfer_contacts(session):
     with open(co_to_org_mapper_path, "r") as f:
         co_to_org_mapper = json.load(f)
 
-    input_df = read_csv("OwnersData")
+    source_table = "OwnersData"
+    input_df = read_csv(source_table)
     odf = input_df.drop(["OBJECTID", "GlobalID"], axis=1)
     ldf = read_csv("OwnerLink")
     ldf = ldf.drop(["OBJECTID", "GlobalID"], axis=1)
@@ -75,11 +76,13 @@ def transfer_contacts(session):
     odf = filter_to_valid_point_ids(session, odf)
     cleaned_df = odf
     errors = []
-    # for i, row in odf.iterrows():
-    for chunk in chunk_by_size(odf, 500):
-        things = (
-            session.query(Thing).filter(Thing.name.in_(chunk.PointID.tolist())).all()
-        )
+    added = []
+    odf = odf.sort_values(by=["PointID"])
+
+    for chunk in chunk_by_size(odf, 100):
+        pointids = chunk.PointID.tolist()
+        logger.info(f"Processing chunk {pointids[0]} to {pointids[-1]}")
+        things = session.query(Thing).filter(Thing.name.in_(pointids)).all()
         for i, row in chunk.iterrows():
             thing = next((thing for thing in things if thing.name == row.PointID), None)
             logger.info(f"Processing PointID: {i} {row.PointID}")
@@ -91,22 +94,26 @@ def transfer_contacts(session):
 
             # TODO: use contact_helper.add_contact
             try:
-                _add_first_contact(session, row, thing, co_to_org_mapper)
-                session.commit()
-                # session.flush()
-                logger.info(f"added first contact for PointID {row.PointID}")
+                if _add_first_contact(session, row, thing, co_to_org_mapper, added):
+                    session.commit()
+                    # session.flush()
+                    logger.info(f"added first contact for PointID {row.PointID}")
             except ValidationError as e:
                 logger.critical(
                     f"Skipping first contact for PointID {row.PointID} due to validation error: {e.errors()}"
                 )
-                session.rollback()
-                errors.append({"pointid": row.PointID, "error": e.errors()})
+                # session.rollback()
+                errors.append(
+                    {"pointid": row.PointID, "error": e, "table": source_table}
+                )
             except Exception as e:
                 logger.critical(
                     f"Skipping first contact for PointID {row.PointID} due to error: {e}"
                 )
                 session.rollback()
-                errors.append({"pointid": row.PointID, "error": e})
+                errors.append(
+                    {"pointid": row.PointID, "error": e, "table": source_table}
+                )
 
             try:
                 if (
@@ -119,27 +126,32 @@ def transfer_contacts(session):
                         f"No second contact info for PointID {row.PointID}, skipping."
                     )
                     continue
-                _add_second_contact(session, row, thing, co_to_org_mapper)
-                session.commit()
-                # session.flush()
-                logger.info(f"added second contact for PointID {row.PointID}")
+                if _add_second_contact(session, row, thing, co_to_org_mapper, added):
+                    session.commit()
+                    # session.flush()
+                    logger.info(f"added second contact for PointID {row.PointID}")
+
             except ValidationError as e:
                 logger.critical(
                     f"Skipping second contact for PointID {row.PointID} due to validation error: {e.errors()}"
                 )
-                session.rollback()
-                errors.append({"pointid": row.PointID, "error": e.errors()})
+                # session.rollback()
+                errors.append(
+                    {"pointid": row.PointID, "error": e, "table": source_table}
+                )
             except Exception as e:
                 logger.critical(
                     f"Skipping second contact for PointID {row.PointID} due to error: {e}"
                 )
                 session.rollback()
-                errors.append({"pointid": row.PointID, "error": e})
+                errors.append(
+                    {"pointid": row.PointID, "error": e, "table": source_table}
+                )
 
     return input_df, cleaned_df, errors
 
 
-def _add_first_contact(session, row, thing, co_to_org_mapper):
+def _add_first_contact(session, row, thing, co_to_org_mapper, added):
     # TODO: extract role from OwnerComment
     # role = extract_owner_role(row.OwnerComment)
     role = "Owner"
@@ -149,6 +161,10 @@ def _add_first_contact(session, row, thing, co_to_org_mapper):
 
     organization = co_to_org_mapper.get(row.Company, row.Company)
 
+    if (name, organization) in added:
+        return
+    added.append((name, organization))
+
     contact_data = {
         "thing_id": thing.id,
         "release_status": release_status,
@@ -232,14 +248,18 @@ def _add_first_contact(session, row, thing, co_to_org_mapper):
         )
         if address:
             contact.addresses.append(address)
+    return True
 
 
-def _add_second_contact(session, row, thing, co_to_org_mapper):
+def _add_second_contact(session, row, thing, co_to_org_mapper, added):
 
     release_status = "private"
     name = _make_name(row.SecondFirstName, row.SecondLastName)
 
     organization = co_to_org_mapper.get(row.Company, row.Company)
+    if (name, organization) in added:
+        return
+    added.append((name, organization))
 
     contact_data = {
         "thing_id": thing.id,
@@ -280,6 +300,7 @@ def _add_second_contact(session, row, thing, co_to_org_mapper):
                 contact.phones.append(phone)
             else:
                 contact.incomplete_nma_phones.append(phone)
+    return True
 
 
 # helpers