BDMS 55: sensor coverage | reusable test functions | PydanticStyleException

api/sensor.py

@@ -14,34 +14,102 @@
 # limitations under the License.
 # ===============================================================================
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Query, Response
 from fastapi_pagination.ext.sqlalchemy import paginate
 from sqlalchemy import select, and_
-from sqlalchemy.orm import Session
 from starlette import status
 
 from api.pagination import CustomPage
 from core.dependencies import session_dependency
 from db import adder, Observation
-from db.engine import get_db_session
 from db.sensor import Sensor
-from schemas.sensor import SensorResponse, CreateSensor
-from services.query_helper import order_sort_filter
+from schemas.sensor import SensorResponse, CreateSensor, UpdateSensor
+from services.crud_helper import model_patcher, model_deleter
+from services.exceptions_helper import PydanticStyleException
+from services.query_helper import order_sort_filter, simple_get_by_id
 
 router = APIRouter(prefix="/sensor", tags=["sensor"])
 
+# ====== POST ==================================================================
+
 
 @router.post("", status_code=status.HTTP_201_CREATED)
 def add_sensor(
-    sensor_data: CreateSensor, session: Session = Depends(get_db_session)
+    sensor_data: CreateSensor, session: session_dependency
 ) -> SensorResponse:
     """
     Add a sensor to the system.
-    This endpoint is a placeholder and should be implemented with actual logic.
     """
     return adder(session, Sensor, sensor_data)
 
 
+# ====== PATCH =================================================================
+
+
+@router.patch("/{sensor_id}", status_code=status.HTTP_200_OK)
+def update_sensor(
+    sensor_id: int, sensor_data: UpdateSensor, session: session_dependency
+) -> SensorResponse:
+    """
+    Update a sensor in the system.
+    """
+    if (
+        sensor_data.datetime_installed is not None
+        and sensor_data.datetime_removed is None
+    ):
+        sensor = simple_get_by_id(session, Sensor, sensor_id)
+        existing_datetime_removed = sensor.datetime_removed
+        if (
+            existing_datetime_removed is not None
+            and sensor_data.datetime_installed >= existing_datetime_removed
+        ):
+            raise PydanticStyleException(
+                status_code=status.HTTP_409_CONFLICT,
+                loc=["body", "datetime_installed"],
+                msg=f"new datetime installed must be before existing datetime removed of {existing_datetime_removed.isoformat().replace('+00:00', 'Z')}",
+                type="value_error",
+                input={
+                    "datetime_installed": sensor_data.datetime_installed.isoformat().replace(
+                        "+00:00", "Z"
+                    )
+                },
+            )
+    elif (
+        sensor_data.datetime_installed is None
+        and sensor_data.datetime_removed is not None
+    ):
+        sensor = simple_get_by_id(session, Sensor, sensor_id)
+        existing_datetime_installed = sensor.datetime_installed
+        if sensor_data.datetime_removed <= existing_datetime_installed:
+            raise PydanticStyleException(
+                status_code=status.HTTP_409_CONFLICT,
+                loc=["body", "datetime_removed"],
+                msg=f"new datetime removed must be after existing datetime installed of {existing_datetime_installed.isoformat().replace('+00:00', 'Z')}",
+                type="value_error",
+                input={
+                    "datetime_removed": sensor_data.datetime_removed.isoformat().replace(
+                        "+00:00", "Z"
+                    )
+                },
+            )
+
+    return model_patcher(session, Sensor, sensor_id, sensor_data)
+
+
+# ====== DELETE ================================================================
+
+
+@router.delete("/{sensor_id}")
+def delete_sensor(sensor_id: int, session: session_dependency) -> Response:
+    """
+    Delete a sensor in the system
+    """
+    return model_deleter(session, Sensor, sensor_id)
+
+
+# ====== GET ===================================================================
+
+
 @router.get("", status_code=status.HTTP_200_OK)
 def get_sensors(
     session: session_dependency,
@@ -56,6 +124,7 @@ def get_sensors(
     This endpoint is a placeholder and should be implemented with actual logic.
     """
     sql = select(Sensor)
+    # TODO: a sensor is not yet related to observation, so this won't work at the moment
     if thing_id is not None or observed_property is not None:
         conditions = []
         if observed_property is not None:
@@ -71,12 +140,11 @@ def get_sensors(
 
 
 @router.get("/{sensor_id}", status_code=status.HTTP_200_OK)
-def get_sensor(
-    sensor_id: int, session: Session = Depends(get_db_session)
-) -> SensorResponse:
-
-    sensor = session.get(Sensor, sensor_id)
-    return sensor
+def get_sensor(sensor_id: int, session: session_dependency) -> SensorResponse:
+    """
+    Retrieve a sensor by its ID.
+    """
+    return simple_get_by_id(session, Sensor, sensor_id)
 
 
 # ============= EOF =============================================

db/__init__.py

@@ -30,8 +30,7 @@
 from db.observation import *
 from db.publication import *
 from db.sample import *
-from db.sensor.groundwaterlevel import *
-from db.sensor.sensor import *
+from db.sensor import *
 from db.thing import *
 
 from sqlalchemy import (

db/sensor/sensor.py → db/sensor.py

@@ -13,23 +13,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ===============================================================================
-from sqlalchemy import Column, String, Integer, ForeignKey, DateTime
-from sqlalchemy.orm import declared_attr, relationship
+from sqlalchemy import Column, String, Integer, DateTime
+from sqlalchemy.orm import relationship
 
 from db.base import Base, AutoBaseMixin
 
 
-class SensorMixin:
-    @declared_attr
-    def sensor_id(self):
-        return Column(
-            Integer,
-            ForeignKey("sensor.id", ondelete="CASCADE"),
-            nullable=False,
-            unique=True,
-        )
-
-
 class Sensor(Base, AutoBaseMixin):
     """
     Base class for all sensor types.
@@ -40,8 +29,8 @@ class Sensor(Base, AutoBaseMixin):
     name = Column(String(255), nullable=False)
     model = Column(String(50))
     serial_no = Column(String(50))
-    date_installed = Column(DateTime)
-    date_removed = Column(DateTime)
+    datetime_installed = Column(DateTime(timezone=True), nullable=False)
+    datetime_removed = Column(DateTime(timezone=True))
     recording_interval = Column(Integer)
     notes = Column(String(50))
 

schemas/sample.py

@@ -51,8 +51,8 @@ class ValidateSample(BaseModel):
     #             )
     #     return sample_bottom
 
-    # REFACTOR TODO: fields are evaluated in the order in which they are defined.
-    # are sample top/bottom really working as expected?
+    sample_top: float | None = None
+    sample_bottom: float | None = None
 
     @model_validator(mode="after")
     def validate_top_and_bottom(self) -> Self:

schemas/sensor.py

@@ -13,13 +13,44 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ===============================================================================
-from datetime import datetime
+from typing_extensions import Annotated, Self
+from datetime import timezone
 
-from pydantic import BaseModel
+from pydantic import (
+    BaseModel,
+    AwareDatetime,
+    PastDatetime,
+    model_validator,
+    field_validator,
+)
+
+# ------- VALIDATION ------
+
+
+class ValidateSensor(BaseModel):
+
+    datetime_installed: AwareDatetime | None = None
+    datetime_removed: AwareDatetime | None = None
+
+    @field_validator("datetime_installed", "datetime_removed")
+    def convert_datetime_fields_to_utc(cls, field: AwareDatetime) -> AwareDatetime:
+        if field is not None and field.tzinfo != timezone.utc:
+            field = field.astimezone(timezone.utc)
+        return field
+
+    @model_validator(mode="after")
+    def check_datetime_values(self) -> Self:
+        if (
+            getattr(self, "datetime_removed", None) is not None
+            and getattr(self, "datetime_installed", None) is not None
+        ):
+            if self.datetime_removed <= self.datetime_installed:
+                raise ValueError("datetime removed must be after datetime installed")
+        return self
 
 
 # -------- CREATE ----------
-class CreateSensor(BaseModel):
+class CreateSensor(ValidateSensor):
     """
     Schema for creating a new sensor.
     """
@@ -28,8 +59,19 @@ class CreateSensor(BaseModel):
     # equipment_type: str | None = None
     model: str | None = None
     serial_no: str | None = None
-    date_installed: str | None = None  # ISO format date string
-    date_removed: str | None = None  # ISO format date string
+    datetime_installed: Annotated[AwareDatetime, PastDatetime()]
+    datetime_removed: AwareDatetime | None = None  # ISO format date string
+    recording_interval: int | None = None
+    notes: str | None = None
+
+
+# -------- UPDATE ----------
+class UpdateSensor(ValidateSensor):
+    name: str | None = None
+    model: str | None = None
+    serial_no: str | None = None
+    datetime_installed: AwareDatetime | None = None
+    datetime_removed: AwareDatetime | None = None
     recording_interval: int | None = None
     notes: str | None = None
 
@@ -40,12 +82,10 @@ class SensorResponse(BaseModel):
     name: str
     model: str | None  # = Column(String(50))
     serial_no: str | None  # = Column(String(50))
-    date_installed: datetime | None  # = Column(DateTime)
-    date_removed: datetime | None  # = Column(DateTime)
+    datetime_installed: AwareDatetime
+    datetime_removed: AwareDatetime | None  # = Column(DateTime)
     recording_interval: int | None  # = Column(Integer)
     notes: str | None  # = Column(String(50))
 
 
-# -------- UPDATE ----------
-
 # ============= EOF =============================================

services/exceptions_helper.py

@@ -0,0 +1,21 @@
+from fastapi import HTTPException
+
+
+class PydanticStyleException(HTTPException):
+    """
+    Exception to be raised for errors not handled by Pydantic to maintain
+    the same style.
+    """
+
+    def __init__(
+        self,
+        status_code: int,
+        loc: list,
+        msg: str,
+        type: str,
+        input: dict,
+    ):
+        super().__init__(
+            status_code=status_code,
+            detail=[{"loc": loc, "msg": msg, "type": type, "input": input}],
+        )

tests/__init__.py

@@ -17,7 +17,7 @@
 
 from core.app import init_lexicon
 from db import Base
-from db.engine import engine
+from db.engine import engine, session_ctx
 from main import app
 
 
@@ -29,4 +29,25 @@
 client = TestClient(app)
 
 
+def cleanup_post_test(model: Base, new_record_id: int) -> None:
+    """
+    Function to cleanup POST tests
+    """
+    with session_ctx() as session:
+        session.delete(session.get(model, new_record_id))
+        session.commit()
+
+
+def cleanup_patch_test(model: Base, payload: dict, original_data: Base) -> None:
+    """
+    Function to cleanup PATCH tests
+    """
+    with session_ctx() as session:
+        updated_record = session.get(model, original_data.id)
+        for field in payload.keys():
+            original_value = getattr(original_data, field)
+            setattr(updated_record, field, original_value)
+        session.commit()
+
+
 # ============= EOF =============================================