API: create /{etype}/{eid}/master and /{etype}/{eid}/snapshots routes

They are just `/{etype}/{eid}` splitted into two parts.

Author: dbnk0

dp3/api/internal/entity_response_models.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Annotated, Any, Optional
 
-from pydantic import BaseModel, Field, NonNegativeInt
+from pydantic import BaseModel, Field, NonNegativeInt, RootModel
 
 from dp3.common.attrspec import AttrSpecType, AttrType
 
@@ -34,6 +34,11 @@ class EntityEidList(BaseModel):
     data: list[dict]
 
 
+EntityEidMasterRecord = RootModel[dict]
+
+EntityEidSnapshots = RootModel[list[dict]]
+
+
 class EntityEidData(BaseModel):
     """Data of entity eid
 

dp3/api/routers/entity.py

@@ -10,6 +10,8 @@
     EntityEidAttrValueOrHistory,
     EntityEidData,
     EntityEidList,
+    EntityEidMasterRecord,
+    EntityEidSnapshots,
 )
 from dp3.api.internal.helpers import api_to_dp3_datapoint
 from dp3.api.internal.models import (
@@ -28,6 +30,43 @@ async def check_etype(etype: str):
     return etype
 
 
+def get_eid_master_record_handler(
+    etype: str, eid: str, date_from: Optional[datetime] = None, date_to: Optional[datetime] = None
+):
+    """Handler for getting master record of EID"""
+    # TODO: This is probably not the most efficient way. Maybe gather only
+    # plain data from master record and then call `get_timeseries_history`
+    # for timeseries.
+    master_record = DB.get_master_record(etype, eid)
+    if "_id" in master_record:
+        del master_record["_id"]
+    if "#hash" in master_record:
+        del master_record["#hash"]
+
+    entity_attribs = MODEL_SPEC.attribs(etype)
+
+    # Get filtered timeseries data
+    for attr in master_record:
+        # Check for no longer existing attributes
+        if attr in entity_attribs and entity_attribs[attr].t == AttrType.TIMESERIES:
+            master_record[attr] = DB.get_timeseries_history(
+                etype, attr, eid, t1=date_from, t2=date_to
+            )
+
+    return master_record
+
+
+def get_eid_snapshots_handler(
+    etype: str, eid: str, date_from: Optional[datetime] = None, date_to: Optional[datetime] = None
+):
+    """Handler for getting snapshots of EID"""
+    snapshots = list(DB.get_snapshots(etype, eid, t1=date_from, t2=date_to))
+    for s in snapshots:
+        del s["_id"]
+
+    return snapshots
+
+
 router = APIRouter(dependencies=[Depends(check_etype)])
 
 # As variable, because otherwise generates ruff B008 error
@@ -104,38 +143,34 @@ async def get_eid_data(
 
     Contains all snapshots and master record.
     Snapshots are ordered by ascending creation time.
-    """
-    # Get master record
-    # TODO: This is probably not the most efficient way. Maybe gather only
-    # plain data from master record and then call `get_timeseries_history`
-    # for timeseries.
-    master_record = DB.get_master_record(etype, eid)
-    if "_id" in master_record:
-        del master_record["_id"]
-    if "#hash" in master_record:
-        del master_record["#hash"]
-
-    entity_attribs = MODEL_SPEC.attribs(etype)
 
-    # Get filtered timeseries data
-    for attr in master_record:
-        # Check for no longer existing attributes
-        if attr in entity_attribs and entity_attribs[attr].t == AttrType.TIMESERIES:
-            master_record[attr] = DB.get_timeseries_history(
-                etype, attr, eid, t1=date_from, t2=date_to
-            )
-
-    # Get snapshots
-    snapshots = list(DB.get_snapshots(etype, eid, t1=date_from, t2=date_to))
-    for s in snapshots:
-        del s["_id"]
+    Combines function of `/{etype}/{eid}/master` and `/{etype}/{eid}/snapshots`.
+    """
+    master_record = get_eid_master_record_handler(etype, eid, date_from, date_to)
+    snapshots = get_eid_snapshots_handler(etype, eid, date_from, date_to)
 
     # Whether this eid contains any data
     empty = not master_record and len(snapshots) == 0
 
     return EntityEidData(empty=empty, master_record=master_record, snapshots=snapshots)
 
 
+@router.get("/{etype}/{eid}/master")
+async def get_eid_master_record(
+    etype: str, eid: str, date_from: Optional[datetime] = None, date_to: Optional[datetime] = None
+) -> EntityEidMasterRecord:
+    """Get master record of `etype`'s `eid`."""
+    return get_eid_master_record_handler(etype, eid, date_from, date_to)
+
+
+@router.get("/{etype}/{eid}/snapshots")
+async def get_eid_snapshots(
+    etype: str, eid: str, date_from: Optional[datetime] = None, date_to: Optional[datetime] = None
+) -> EntityEidSnapshots:
+    """Get snapshots of `etype`'s `eid`."""
+    return get_eid_snapshots_handler(etype, eid, date_from, date_to)
+
+
 @router.get("/{etype}/{eid}/get/{attr}")
 async def get_eid_attr_value(
     etype: str,

tests/test_api/test_get_entity_eid_data.py

@@ -3,7 +3,7 @@
 
 import common
 
-from dp3.api.internal.entity_response_models import EntityEidData
+from dp3.api.internal.entity_response_models import EntityEidData, EntityEidMasterRecord
 
 DATAPOINT_COUNT = 6
 
@@ -44,3 +44,12 @@ def test_get_entity_data(self):
             lambda data: "test_attr_history" in data.master_record,
         )
         self.assertEqual(DATAPOINT_COUNT, len(data.master_record["test_attr_history"]))
+
+    def test_get_entity_master_record(self):
+        data = self.query_expected_value(
+            lambda: self.get_entity_data(
+                f"entity/test_entity_type/{self.eid}/master", EntityEidMasterRecord
+            ),
+            lambda data: "test_attr_history" in data.root,
+        )
+        self.assertEqual(DATAPOINT_COUNT, len(data.root["test_attr_history"]))