feat: add datastore API for fetching a single event by ID

aw_datastore/datastore.py

@@ -102,6 +102,11 @@ def get(
             self.bucket_id, limit, starttime, endtime
         )
 
+    def get_by_id(self, event_id) -> Event:
+        return self.ds.storage_strategy.get_event(
+            self.bucket_id, event_id
+        )
+
     def get_eventcount(
         self, starttime: datetime = None, endtime: datetime = None
     ) -> int:

aw_datastore/storages/abstract.py

@@ -42,6 +42,14 @@ def delete_bucket(self, bucket_id: str) -> None:
     def get_metadata(self, bucket_id: str) -> dict:
         raise NotImplementedError
 
+    @abstractmethod
+    def get_event(
+        self,
+        bucket_id: str,
+        event_id: int,
+    ) -> List[Event]:
+        raise NotImplementedError
+
     @abstractmethod
     def get_events(
         self,

aw_datastore/storages/memory.py

@@ -49,6 +49,14 @@ def buckets(self):
             buckets[bucket_id] = self.get_metadata(bucket_id)
         return buckets
 
+    def get_event(
+        self,
+        bucket_id: str,
+        event_id: int,
+    ) -> List[Event]:
+        event = self._get_event(bucket_id, event_id)
+        return copy.deepcopy(event)
+
     def get_events(
         self,
         bucket: str,
@@ -117,6 +125,11 @@ def delete(self, bucket_id, event_id):
             return True
         return False
 
+    def _get_event(self, bucket_id, event_id):
+        events = [event for idx, event in reversed(list(enumerate(self.db[bucket_id]))) if event.id == event_id]
+        assert len(events) == 1
+        return events[0]
+
     def replace(self, bucket_id, event_id, event):
         for idx in (
             idx

aw_datastore/storages/peewee.py

@@ -254,6 +254,17 @@ def replace(self, bucket_id, event_id, event):
         event.id = e.id
         return event
 
+    def get_event(
+        self,
+        bucket_id: str,
+        event_id: int,
+    ):
+        """
+        Fetch a single event from a bucket.
+        """
+        res = self._get_event(bucket_id, event_id)
+        return Event(**EventModel.json(res))
+
     def get_events(
         self,
         bucket_id: str,

aw_datastore/storages/sqlite.py

@@ -54,6 +54,20 @@
 """
 
 
+def _rows_to_events(rows: list) -> Event:
+    events = []
+    for row in rows:
+        eid = row[0]
+        starttime = datetime.fromtimestamp(row[1] / 1000000, timezone.utc)
+        endtime = datetime.fromtimestamp(row[2] / 1000000, timezone.utc)
+        duration = endtime - starttime
+        data = json.loads(row[3])
+        events.append(
+            Event(id=eid, timestamp=starttime, duration=duration, data=data)
+        )
+    return events
+
+
 class SqliteStorage(AbstractStorage):
     sid = "sqlite"
 
@@ -256,6 +270,23 @@ def replace(self, bucket_id, event_id, event) -> bool:
         self.conditional_commit(1)
         return True
 
+    def get_event(
+        self,
+        bucket_id: str,
+        event_id: str,
+    ):
+        self.commit()
+        c = self.conn.cursor()
+        query = """
+            SELECT id, starttime, endtime, datastr
+            FROM events
+            WHERE bucketrow = (SELECT rowid FROM buckets WHERE id = ?) AND id = ?
+            LIMIT 1
+        """
+        rows = c.execute(query, [bucket_id, event_id])
+        events = _rows_to_events(rows)
+        return events[0]
+
     def get_events(
         self,
         bucket_id: str,
@@ -279,16 +310,7 @@ def get_events(
             ORDER BY endtime DESC LIMIT ?
         """
         rows = c.execute(query, [bucket_id, starttime_i, endtime_i, limit])
-        events = []
-        for row in rows:
-            eid = row[0]
-            starttime = datetime.fromtimestamp(row[1] / 1000000, timezone.utc)
-            endtime = datetime.fromtimestamp(row[2] / 1000000, timezone.utc)
-            duration = endtime - starttime
-            data = json.loads(row[3])
-            events.append(
-                Event(id=eid, timestamp=starttime, duration=duration, data=data)
-            )
+        events = _rows_to_events(rows)
         return events
 
     def get_eventcount(

tests/test_datastore.py

@@ -286,6 +286,25 @@ def test_get_datefilter_simple(bucket_cm):
         assert 1 == len(fetched_events)
 
 
+@pytest.mark.parametrize("bucket_cm", param_testing_buckets_cm())
+def test_get_event_by_id(bucket_cm):
+    """Test that we can retrieve single events by their IDs"""
+    with bucket_cm as bucket:
+        eventcount = 2
+        # Create 1-day long events
+        events = [
+            Event(timestamp=now + i * td1d, duration=td1d) for i in range(eventcount)
+        ]
+        bucket.insert(events)
+
+        # Retrieve stored events
+        events = bucket.get()
+        for e in events:
+            # Query them one-by-one
+            event = bucket.get_by_id(e.id)
+            assert e == event
+
+
 @pytest.mark.parametrize("bucket_cm", param_testing_buckets_cm())
 def test_get_event_trimming(bucket_cm):
     """Test that event trimming works correctly (when querying events that intersect with the query range)"""