diff --git a/api/thing.py b/api/thing.py index a4ebab305..907bd8470 100644 --- a/api/thing.py +++ b/api/thing.py @@ -33,6 +33,7 @@ viewer_dependency, ) from db.thing import Thing, ThingIdLink, WellScreen +from db.deployment import Deployment from schemas.thing import ( CreateThingIdLink, CreateWell, @@ -48,6 +49,7 @@ UpdateThingIdLink, UpdateWellScreen, ) +from schemas.deployment import DeploymentResponse from services.crud_helper import model_patcher, model_adder, model_deleter from services.exceptions_helper import PydanticStyleException from services.query_helper import ( @@ -345,6 +347,20 @@ async def get_thing_id_links( return paginate(query=sql, conn=session) +@router.get("/{thing_id}/deployment", summary="Get deployments by thing ID") +async def get_thing_deployments( + user: viewer_dependency, + thing_id: int, + session: session_dependency, +) -> CustomPage[DeploymentResponse]: + """ + Retrieve all deployments for a specific thing by its ID. + """ + thing = simple_get_by_id(session, Thing, thing_id) + sql = select(Deployment).where(Deployment.thing_id == thing.id) + return paginate(query=sql, conn=session) + + # POST ======================================================================== diff --git a/db/deployment.py b/db/deployment.py index 0b2dc61df..8f4aaf84e 100644 --- a/db/deployment.py +++ b/db/deployment.py @@ -50,4 +50,6 @@ class Deployment(Base, AutoBaseMixin, ReleaseMixin): # Many-To-One: A Deployment is for one Thing. thing: Mapped["Thing"] = relationship("Thing", back_populates="deployments") # Many-To-One: A Deployment is of one piece of equipment (sensor). - sensor: Mapped["Sensor"] = relationship("Sensor", back_populates="deployments") + sensor: Mapped["Sensor"] = relationship( + "Sensor", back_populates="deployments", lazy="joined" + ) diff --git a/schemas/deployment.py b/schemas/deployment.py new file mode 100644 index 000000000..5bd050145 --- /dev/null +++ b/schemas/deployment.py @@ -0,0 +1,17 @@ +from datetime import date + +from schemas import BaseResponseModel +from schemas.sensor import SensorResponse + + +class DeploymentResponse(BaseResponseModel): + thing_id: int + sensor: SensorResponse + installation_date: date + removal_date: date | None + recording_interval: int | None + recording_interval_units: str | None + hanging_cable_length: float | None + hanging_point_height: float | None + hanging_point_description: str | None + notes: str | None diff --git a/tests/test_thing.py b/tests/test_thing.py index 12a728b2b..f94defc37 100644 --- a/tests/test_thing.py +++ b/tests/test_thing.py @@ -718,6 +718,47 @@ def test_get_thing_by_id_404_not_found(water_well_thing): assert data["detail"] == f"Thing with ID {bad_id} not found." +def test_get_thing_deployments_by_id( + water_well_thing, sensor_to_water_well_thing_deployment, sensor +): + response = client.get(f"/thing/{water_well_thing.id}/deployment") + assert response.status_code == 200 + data = response.json() + assert data["total"] == 1 + assert data["items"][0]["id"] == sensor_to_water_well_thing_deployment.id + assert data["items"][0]["thing_id"] == water_well_thing.id + assert data["items"][0]["sensor"]["id"] == sensor.id + assert ( + data["items"][0]["installation_date"] + == sensor_to_water_well_thing_deployment.installation_date + ) + assert ( + data["items"][0]["removal_date"] + == sensor_to_water_well_thing_deployment.removal_date + ) + assert ( + data["items"][0]["recording_interval"] + == sensor_to_water_well_thing_deployment.recording_interval + ) + assert ( + data["items"][0]["recording_interval_units"] + == sensor_to_water_well_thing_deployment.recording_interval_units + ) + assert ( + data["items"][0]["hanging_cable_length"] + == sensor_to_water_well_thing_deployment.hanging_cable_length + ) + assert ( + data["items"][0]["hanging_point_height"] + == sensor_to_water_well_thing_deployment.hanging_point_height + ) + assert ( + data["items"][0]["hanging_point_description"] + == sensor_to_water_well_thing_deployment.hanging_point_description + ) + assert data["items"][0]["notes"] == sensor_to_water_well_thing_deployment.notes + + # # weaver tests # def test_weaver_get_wells_geojson(): # response = client.get("/geospatial", params={"type": "well"})