Create object history API (#2074)

Signed-off-by: Donny Peeters <[email protected]>
Co-authored-by: ammar92 <[email protected]>
Co-authored-by: Jan Klopper <[email protected]>

octopoes/octopoes/api/router.py

@@ -33,6 +33,7 @@
 from octopoes.models.origin import Origin, OriginParameter, OriginType
 from octopoes.models.pagination import Paginated
 from octopoes.models.path import Path as ObjectPath
+from octopoes.models.transaction import TransactionRecord
 from octopoes.models.tree import ReferenceTree
 from octopoes.models.types import type_by_name
 from octopoes.version import __version__
@@ -170,6 +171,28 @@ def get_object(
     return octopoes.get_ooi(reference, valid_time)
 
 
+@router.get("/object-history", tags=["Objects"])
+def get_object_history(
+    reference: Reference = Depends(extract_reference),
+    sort_order: str = "asc",  # Or: "desc"
+    with_docs: bool = False,
+    has_doc: Optional[bool] = None,
+    offset: int = 0,
+    limit: Optional[int] = None,
+    indices: Optional[List[int]] = None,
+    octopoes: OctopoesService = Depends(octopoes_service),
+) -> List[TransactionRecord]:
+    return octopoes.get_ooi_history(
+        reference,
+        sort_order=sort_order,
+        with_docs=with_docs,
+        has_doc=has_doc,
+        offset=offset,
+        limit=limit,
+        indices=indices,
+    )
+
+
 @router.get("/objects/random", tags=["Objects"])
 def list_random_objects(
     octopoes: OctopoesService = Depends(octopoes_service),

octopoes/octopoes/connector/octopoes.py

@@ -27,6 +27,7 @@
 from octopoes.models.ooi.findings import Finding, RiskLevelSeverity
 from octopoes.models.origin import Origin, OriginParameter, OriginType
 from octopoes.models.pagination import Paginated
+from octopoes.models.transaction import TransactionRecord
 from octopoes.models.tree import ReferenceTree
 from octopoes.models.types import OOIType
 
@@ -114,6 +115,31 @@ def get(self, reference: Reference, valid_time: Optional[datetime] = None) -> OO
         )
         return TypeAdapter(OOIType).validate_json(res.content)
 
+    def get_history(
+        self,
+        reference: Reference,
+        *,
+        sort_order: str = "asc",  # Or: "desc"
+        with_docs: bool = False,
+        has_doc: Optional[bool] = None,
+        offset: int = 0,
+        limit: Optional[int] = None,
+        indices: Optional[List[int]] = None,
+    ) -> List[TransactionRecord]:
+        res = self.session.get(
+            f"/{self.client}/object-history",
+            params={
+                "reference": str(reference),
+                "sort_order": sort_order,
+                "with_docs": with_docs,
+                "has_doc": has_doc,
+                "offset": offset,
+                "limit": limit,
+                "indices": indices,
+            },
+        )
+        return TypeAdapter(List[TransactionRecord]).validate_json(res.content)
+
     def get_tree(
         self,
         reference: Reference,

octopoes/octopoes/core/service.py

@@ -41,6 +41,7 @@
     get_max_scan_level_issuance,
     get_paths_to_neighours,
 )
+from octopoes.models.transaction import TransactionRecord
 from octopoes.models.tree import ReferenceTree
 from octopoes.repositories.ooi_repository import OOIRepository
 from octopoes.repositories.origin_parameter_repository import OriginParameterRepository
@@ -92,6 +93,27 @@ def get_ooi(self, reference: Reference, valid_time: datetime) -> OOI:
         ooi = self.ooi_repository.get(reference, valid_time)
         return self._populate_scan_profiles([ooi], valid_time)[0]
 
+    def get_ooi_history(
+        self,
+        reference: Reference,
+        *,
+        sort_order: str = "asc",  # Or: "desc"
+        with_docs: bool = False,
+        has_doc: Optional[bool] = None,
+        offset: int = 0,
+        limit: Optional[int] = None,
+        indices: Optional[List[int]] = None,
+    ) -> List[TransactionRecord]:
+        return self.ooi_repository.get_history(
+            reference,
+            sort_order=sort_order,
+            with_docs=with_docs,
+            has_doc=has_doc,
+            offset=offset,
+            limit=limit,
+            indices=indices,
+        )
+
     def list_ooi(
         self,
         types: Set[Type[OOI]],

octopoes/octopoes/models/transaction.py

@@ -0,0 +1,12 @@
+from datetime import datetime
+from typing import Dict, Optional
+
+from pydantic import BaseModel, Field
+
+
+class TransactionRecord(BaseModel):
+    transaction_time: datetime = Field(alias="txTime")
+    transaction_id: int = Field(alias="txId")
+    valid_time: datetime = Field(alias="validTime")
+    content_hash: str = Field(alias="contentHash")
+    document: Optional[Dict] = Field(None, alias="doc")

octopoes/octopoes/repositories/ooi_repository.py

@@ -31,6 +31,7 @@
 from octopoes.models.ooi.findings import Finding, FindingType, RiskLevelSeverity
 from octopoes.models.pagination import Paginated
 from octopoes.models.path import Direction, Path, Segment, get_paths_to_neighours
+from octopoes.models.transaction import TransactionRecord
 from octopoes.models.tree import ReferenceNode, ReferenceTree
 from octopoes.models.types import get_concrete_types, get_relation, get_relations, to_concrete, type_by_name
 from octopoes.repositories.repository import Repository
@@ -68,6 +69,19 @@ def __init__(self, event_manager: EventManager):
     def get(self, reference: Reference, valid_time: datetime) -> OOI:
         raise NotImplementedError
 
+    def get_history(
+        self,
+        reference: Reference,
+        *,
+        sort_order: str = "asc",  # Or: "desc"
+        with_docs: bool = False,
+        has_doc: Optional[bool] = None,
+        offset: int = 0,
+        limit: Optional[int] = None,
+        indices: Optional[List[int]] = None,
+    ) -> List[TransactionRecord]:
+        raise NotImplementedError
+
     def load_bulk(self, references: Set[Reference], valid_time: datetime) -> Dict[str, OOI]:
         raise NotImplementedError
 
@@ -226,6 +240,31 @@ def get(self, reference: Reference, valid_time: datetime) -> OOI:
             if e.response.status_code == HTTPStatus.NOT_FOUND:
                 raise ObjectNotFoundException(str(reference))
 
+    def get_history(
+        self,
+        reference: Reference,
+        *,
+        sort_order: str = "asc",  # Or: "desc"
+        with_docs: bool = False,
+        has_doc: Optional[bool] = None,
+        offset: int = 0,
+        limit: Optional[int] = None,
+        indices: Optional[List[int]] = None,
+    ) -> List[TransactionRecord]:
+        try:
+            return self.session.client.get_entity_history(
+                str(reference),
+                sort_order=sort_order,
+                with_docs=with_docs,
+                has_doc=has_doc,
+                offset=offset,
+                limit=limit,
+                indices=indices,
+            )
+        except HTTPError as e:
+            if e.response.status_code == HTTPStatus.NOT_FOUND:
+                raise ObjectNotFoundException(str(reference))
+
     def load_bulk(self, references: Set[Reference], valid_time: datetime) -> Dict[str, OOI]:
         ids = list(map(str, references))
         query = generate_pull_query(self.xtdb_type, FieldSet.ALL_FIELDS, {self.pk_prefix(): ids})

octopoes/octopoes/xtdb/client.py

@@ -6,9 +6,10 @@
 from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
 
 import requests
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
 from requests import HTTPError, Response
 
+from octopoes.models.transaction import TransactionRecord
 from octopoes.xtdb.exceptions import NodeNotFound, NoMultinode, XTDBException
 from octopoes.xtdb.query import Query
 
@@ -94,6 +95,39 @@ def get_entity(self, entity_id: str, valid_time: Optional[datetime] = None) -> d
         self._verify_response(res)
         return res.json()
 
+    def get_entity_history(
+        self,
+        entity_id: str,
+        *,
+        sort_order: str = "asc",  # Or: "desc"
+        with_docs: bool = False,
+        has_doc: Optional[bool] = None,
+        offset: int = 0,
+        limit: Optional[int] = None,
+        indices: Optional[List[int]] = None,
+    ) -> List[TransactionRecord]:
+        params = {
+            "eid": entity_id,
+            "sort-order": sort_order,
+            "history": "true",
+            "with-docs": "true" if with_docs else "false",
+        }
+
+        res = self._session.get(f"{self.client_url()}/entity", params=params)
+        self._verify_response(res)
+        transactions: List[TransactionRecord] = TypeAdapter(List[TransactionRecord]).validate_json(res.content)
+
+        if has_doc is True:  # The doc is None if and only if the hash is  "0000000000000000000000000000000000000000"
+            transactions = [transaction for transaction in transactions if transaction.content_hash != 40 * "0"]
+
+        if has_doc is False:  # The doc is None if and only if the hash is  "0000000000000000000000000000000000000000"
+            transactions = [transaction for transaction in transactions if transaction.content_hash == 40 * "0"]
+
+        if indices:
+            return [tx for i, tx in enumerate(transactions) if i in indices or i - len(transactions) in indices]
+
+        return transactions[offset:limit]
+
     def query(self, query: Union[str, Query], valid_time: Optional[datetime] = None) -> List[List[Any]]:
         if valid_time is None:
             valid_time = datetime.now(timezone.utc)

octopoes/tests/integration/test_api_connector.py

@@ -1,6 +1,6 @@
 import os
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 from ipaddress import ip_address
 from typing import List
 
@@ -72,6 +72,45 @@ def test_bulk_operations(octopoes_api_connector: OctopoesAPIConnector, valid_tim
     assert octopoes_api_connector.list(types={Network, Hostname}).count == 6
 
 
+def test_history(octopoes_api_connector: OctopoesAPIConnector):
+    network = Network(name="test")
+    first_seen = datetime(year=2020, month=10, day=10, tzinfo=timezone.utc)  # XTDB only returns a precision of seconds
+    octopoes_api_connector.save_declaration(
+        Declaration(
+            ooi=network,
+            valid_time=first_seen,
+        )
+    )
+    octopoes_api_connector.delete(network.reference, datetime(year=2020, month=10, day=11, tzinfo=timezone.utc))
+    last_seen = datetime(year=2020, month=10, day=12, tzinfo=timezone.utc)
+    octopoes_api_connector.save_declaration(
+        Declaration(
+            ooi=network,
+            valid_time=last_seen,
+        )
+    )
+
+    history = octopoes_api_connector.get_history(network.reference, with_docs=True)
+    assert len(history) == 3
+    assert history[0].document is not None
+    assert history[1].document is None
+    assert history[2].document is not None
+
+    assert len(octopoes_api_connector.get_history(network.reference, has_doc=False)) == 1
+
+    with_doc = octopoes_api_connector.get_history(network.reference, has_doc=True)
+    assert len(with_doc) == 2
+    assert not all([x.document for x in with_doc])
+
+    assert len(octopoes_api_connector.get_history(network.reference, offset=1)) == 2
+    assert len(octopoes_api_connector.get_history(network.reference, limit=2)) == 2
+
+    first_and_last = octopoes_api_connector.get_history(network.reference, has_doc=True, indices=[1, -1])
+    assert len(first_and_last) == 2
+    assert first_and_last[0].valid_time == first_seen
+    assert first_and_last[1].valid_time == last_seen
+
+
 def test_query(octopoes_api_connector: OctopoesAPIConnector, valid_time: datetime):
     network = Network(name="test")
     octopoes_api_connector.save_declaration(

octopoes/tests/integration/test_xtdb_client.py

@@ -1,5 +1,5 @@
 import os
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytest
 from requests import HTTPError
@@ -11,7 +11,7 @@
 from octopoes.models.ooi.network import Network
 from octopoes.models.path import Path
 from octopoes.repositories.ooi_repository import XTDBOOIRepository
-from octopoes.xtdb.client import XTDBHTTPClient, XTDBSession
+from octopoes.xtdb.client import OperationType, XTDBHTTPClient, XTDBSession
 from octopoes.xtdb.exceptions import NodeNotFound
 from octopoes.xtdb.query import Query
 from tests.conftest import seed_system
@@ -148,6 +148,25 @@ def test_query_empty_on_reference_filter_for_wrong_hostname(xtdb_session: XTDBSe
     assert len(xtdb_session.client.query(str(Query(Network)))) == 2
 
 
+def test_entity_history(xtdb_session: XTDBSession, valid_time: datetime):
+    network = Network(name="testnetwork")
+    xtdb_session.put(XTDBOOIRepository.serialize(network), datetime.now(timezone.utc))
+    xtdb_session.commit()
+
+    xtdb_session.add((OperationType.DELETE, str(network.reference), datetime.now(timezone.utc)))
+    xtdb_session.commit()
+
+    xtdb_session.put(XTDBOOIRepository.serialize(network), datetime.now(timezone.utc))
+    xtdb_session.commit()
+
+    history = xtdb_session.client.get_entity_history(str(network.reference), with_docs=True)
+    assert len(history) == 3
+
+    assert history[0].document is not None
+    assert history[1].document is None
+    assert history[2].document is not None
+
+
 @pytest.mark.xfail(reason="race condition")
 def test_query_for_system_report(octopoes_api_connector: OctopoesAPIConnector, xtdb_session: XTDBSession, valid_time):
     seed_system(octopoes_api_connector, valid_time)