Updated observations endpoint for testing

Author: Ken Lippold

etl/services/hydroshare_archival.py

@@ -295,7 +295,5 @@ def run(self, user: Optional[User], uid: uuid.UUID, make_public=False):
             "thing_id": str(uid),
             "link": str(thing_archive.settings["link"]),
             "path": thing_archive.settings["path"],
-            "datastreamIds": thing_archive.settings[
-                "datastreamIds"
-            ],
+            "datastreamIds": thing_archive.settings["datastreamIds"],
         }

hydroserver/settings.py

@@ -140,7 +140,7 @@
                 "max_size": config("DB_POOL_MAX_SIZE", default=10, cast=int),
                 "timeout": config("DB_POOL_TIMEOUT", default=60, cast=int),
             },
-            **dj_database_config.get("OPTIONS", {})
+            **dj_database_config.get("OPTIONS", {}),
         },
     }
 }

sta/models/observation.py

@@ -96,7 +96,7 @@ def escape_pg_copy(value):
             ) as copy:
                 buffer = io.StringIO()
                 for i in range(0, len(observations), batch_size):
-                    batch = observations[i: i + batch_size]
+                    batch = observations[i : i + batch_size]
                     buffer.write(
                         "\n".join(
                             "\t".join(
@@ -152,6 +152,4 @@ def get_user_permissions(
         return user_permissions
 
     class Meta:
-        indexes = [
-            models.Index(fields=["id"])
-        ]
+        indexes = [models.Index(fields=["id"])]

sta/schemas/__init__.py

@@ -18,4 +18,9 @@
 )
 from .sensor import SensorGetResponse, SensorPostBody, SensorPatchBody
 from .unit import UnitGetResponse, UnitPostBody, UnitPatchBody
-from .datastream import DatastreamGetResponse, DatastreamPostBody, DatastreamPatchBody
+from .datastream import (
+    DatastreamGetResponse,
+    DatastreamPostBody,
+    DatastreamPatchBody,
+    ObservationsGetResponse,
+)

sta/schemas/datastream.py

@@ -48,3 +48,8 @@ class DatastreamPostBody(BasePostBody, DatastreamFields):
 
 class DatastreamPatchBody(BasePatchBody, DatastreamFields):
     pass
+
+
+class ObservationsGetResponse(BaseGetResponse):
+    phenomenon_time: list
+    result: list

sta/services/datastream.py

@@ -1,6 +1,7 @@
 import uuid
 from typing import Optional, Literal
 from ninja.errors import HttpError
+from datetime import datetime
 from django.contrib.auth import get_user_model
 from django.utils import timezone
 from django.http import StreamingHttpResponse
@@ -358,7 +359,43 @@ def get_csv(self, user: User, uid: uuid.UUID):
 
         return response
 
-    def list_observations(self, user: User, uid: uuid.UUID, ):
+    def list_observations(
+        self,
+        user: User,
+        uid: uuid.UUID,
+        phenomenon_start_time: Optional[datetime] = None,
+        phenomenon_end_time: Optional[datetime] = None,
+        page: int = 1,
+        page_size: Optional[int] = None,
+        order: Literal["asc", "desc"] = "desc",
+    ):
         datastream = self.get_datastream_for_action(user=user, uid=uid, action="view")
 
-        return Observation.objects.filter(datastream=datastream).values_list("phenomenon_time", "result")
+        fields = ["phenomenon_time", "result"]
+
+        queryset = Observation.objects.filter(datastream=datastream)
+        ordering = "phenomenon_time" if order == "asc" else "-phenomenon_time"
+
+        if phenomenon_start_time:
+            queryset = queryset.filter(phenomenon_time__gte=phenomenon_start_time)
+        if phenomenon_end_time:
+            queryset = queryset.filter(phenomenon_time__lte=phenomenon_end_time)
+
+        queryset = queryset.order_by(ordering)
+
+        if page_size is not None:
+            page = max(1, page)
+            page_size = max(0, page_size)
+            start = (page - 1) * page_size
+            end = start + page_size
+
+            queryset = queryset[start:end]
+
+        observations = list(queryset.values_list(*fields))
+
+        if observations:
+            response = dict(zip(fields, zip(*observations)))
+        else:
+            response = {field: [] for field in fields}
+
+        return response

sta/views/datastream.py

@@ -1,10 +1,16 @@
 import uuid
 from ninja import Router, Path
-from typing import Optional
+from typing import Optional, Literal
+from datetime import datetime
 from django.db import transaction
 from hydroserver.security import bearer_auth, session_auth, anonymous_auth
 from hydroserver.http import HydroServerHttpRequest
-from sta.schemas import DatastreamGetResponse, DatastreamPostBody, DatastreamPatchBody
+from sta.schemas import (
+    DatastreamGetResponse,
+    DatastreamPostBody,
+    DatastreamPatchBody,
+    ObservationsGetResponse,
+)
 from sta.services import DatastreamService
 
 datastream_router = Router(tags=["Datastreams"])
@@ -142,13 +148,27 @@ def get_datastream_csv(request: HydroServerHttpRequest, datastream_id: Path[uuid
 @datastream_router.get(
     "/{datastream_id}/observations",
     auth=[session_auth, bearer_auth, anonymous_auth],
-    response={200: list[list], 403: str, 404: str},
+    response={200: ObservationsGetResponse, 403: str, 404: str},
 )
-def get_datastream_observations(request: HydroServerHttpRequest, datastream_id: Path[uuid.UUID]):
+def get_datastream_observations(
+    request: HydroServerHttpRequest,
+    datastream_id: Path[uuid.UUID],
+    phenomenon_start_time: Optional[datetime] = None,
+    phenomenon_end_time: Optional[datetime] = None,
+    page: int = 1,
+    page_size: Optional[int] = None,
+    order: Literal["asc", "desc"] = "desc",
+):
     """
     Get Datastream Observations
     """
 
     return datastream_service.list_observations(
-        user=request.authenticated_user, uid=datastream_id
+        user=request.authenticated_user,
+        uid=datastream_id,
+        phenomenon_start_time=phenomenon_start_time,
+        phenomenon_end_time=phenomenon_end_time,
+        page=page,
+        page_size=page_size,
+        order=order,
     )