Merge pull request #281 from FZJ-INM1-BDA/chore_removeLiveQKgV1

chore: remove live query of kg v1

siibra/configuration/factory.py

@@ -68,6 +68,9 @@ def extract_datasets(cls, spec):
             result.append(
                 datasets.EbrainsDataset(id=spec["ebrains"]["minds/core/dataset/v1.0.0"])
             )
+        if "openminds/Dataset" in spec.get("ebrains", {}):
+            # TODO add parser for ebrains kg v3 dataset. see EbrainsV3Dataset 
+            pass
         return result
 
     @classmethod

siibra/features/dataset/ebrains.py

@@ -64,7 +64,7 @@ def description(self):
 
     @property
     def name(self):
-        return self._name_cached
+        return self._name
 
     @property
     def version_history(self):

siibra/livequeries/ebrains.py

@@ -48,11 +48,6 @@ class EbrainsFeatureQuery(query.LiveQuery, args=[], FeatureType=_ebrains.Ebrains
             requests.GitlabProxy(
                 flavour=requests.GitlabProxyEnum.PARCELLATIONREGION_V1,
             ),
-            requests.EbrainsKgQuery(
-                query_id="siibra-kg-feature-summary-0_0_4",
-                schema="parcellationregion",
-                params={"vocab": "https://schema.hbp.eu/myQuery/"},
-            )
         ]
     )
 

siibra/retrieval/__init__.py

@@ -21,6 +21,6 @@
     LocalFileRepository,
     ZipfileConnector
 )
-from .requests import HttpRequest, ZipfileRequest, EbrainsRequest, EbrainsKgQuery, SiibraHttpRequestError
+from .requests import HttpRequest, ZipfileRequest, EbrainsRequest, SiibraHttpRequestError
 from .cache import CACHE
 from .exceptions import NoSiibraConfigMirrorsAvailableException, TagNotFoundException

siibra/retrieval/datasets.py

@@ -14,10 +14,11 @@
 # limitations under the License.
 
 
-from .requests import EbrainsKgQuery, MultiSourcedRequest, GitlabProxy, GitlabProxyEnum
+from .requests import MultiSourcedRequest, GitlabProxy, GitlabProxyEnum
 
 import re
 from typing import Union, List
+from abc import ABC, abstractproperty
 
 try:
     from typing import TypedDict
@@ -44,15 +45,69 @@ class EbrainsDatasetUrl(TypedDict):
     'identifier': List[str]
 })
 
+class EbrainsBaseDataset(ABC):
+
+    @abstractproperty
+    def id(self) -> str:
+        raise NotImplementedError
+
+    @abstractproperty
+    def name(self) -> str:
+        raise NotImplementedError
+
+    @abstractproperty
+    def urls(self) -> List[EbrainsDatasetUrl]:
+        raise NotImplementedError
+
+    @abstractproperty
+    def description(self) -> str:
+        raise NotImplementedError
+
+    @abstractproperty
+    def contributors(self) -> List[EbrainsDatasetPerson]:
+        raise NotImplementedError
+
+    @abstractproperty
+    def ebrains_page(self) -> str:
+        raise NotImplementedError
+
+    @abstractproperty
+    def custodians(self) -> EbrainsDatasetPerson:
+        raise NotImplementedError
+
+    def __hash__(self):
+        return hash(self.id)
+
+    def __eq__(self, o: object) -> bool:
+        return hasattr(o, "id") and self.id == o.id
+
+    def match(self, spec: Union[str, 'EbrainsBaseDataset']) -> bool:
+        """
+        Checks if the given specification describes this dataset.
+
+        Parameters
+        ----------
+        spec (str, EbrainsBaseDataset)
+            specification to be matched.
+        Returns
+        -------
+        bool
+        """
+        if spec is self:
+            return True
+        if isinstance(spec, str):
+            return self.id == spec
+        raise RuntimeError(f"Cannot match {spec.__class__}, must be either str or EbrainsBaseDataset")
 
-class EbrainsDataset:
+class EbrainsDataset(EbrainsBaseDataset):
 
     def __init__(self, id, name=None, embargo_status: List[EbrainsDatasetEmbargoStatus] = None, *, cached_data=None):
+        super().__init__()
 
         self._id = id
+        self._name = name
         self._cached_data = cached_data
         self.embargo_status = embargo_status
-        self._name_cached = name
 
         if id is None:
             raise TypeError("Dataset id is required")
@@ -78,20 +133,15 @@ def detail(self):
                         GitlabProxyEnum.DATASET_V1,
                         instance_id=instance_id,
                     ),
-                    EbrainsKgQuery(
-                        query_id="interactiveViewerKgQuery-v1_0",
-                        instance_id=instance_id,
-                        params={"vocab": "https://schema.hbp.eu/myQuery/"},
-                    )
                 ]
             ).data
         return self._cached_data
 
     @property
     def name(self) -> str:
-        if self._name_cached is None:
-            self._name_cached = self.detail.get("name")
-        return self._name_cached
+        if self._name is None:
+            self._name = self.detail.get("name")
+        return self._name
 
     @property
     def urls(self) -> List[EbrainsDatasetUrl]:
@@ -118,32 +168,26 @@ def ebrains_page(self):
     def custodians(self) -> EbrainsDatasetPerson:
         return self.detail.get("custodians")
 
-    @property
-    def key(self):
-        return self.id
-
-    def __hash__(self):
-        return hash(self.id)
+class EbrainsV3Dataset(EbrainsBaseDataset):
+    # TODO finish implementing me
+    # some fields are currently missing, e.g. desc, contributors etc.
+    def __init__(self, id, *, cached_data) -> None:
+        super().__init__()
 
-    def __eq__(self, o: object) -> bool:
-        if type(o) is not EbrainsDataset and not issubclass(type(o), EbrainsDataset):
-            return False
-        return self.id == o.id
-
-    def match(self, spec: Union[str, 'EbrainsDataset']) -> bool:
-        """
-        Checks if the given specification describes this dataset.
-
-        Parameters
-        ----------
-        spec (str, EbrainsDataset)
-            specification to be matched.
-        Returns
-        -------
-        bool
-        """
-        if spec is self:
-            return True
-        if isinstance(spec, str):
-            return self.id == spec
-        raise RuntimeError(f"Cannot match {spec.__class__}, must be either str or EbrainsDataset")
+        self._id = id
+        self._cached_data = cached_data
+
+    @property
+    def detail(self):
+        if not self._cached_data:
+            match = re.search(r"([a-f0-9-]+)$", self.id)
+            instance_id = match.group(1)
+            self._cached_data = MultiSourcedRequest(
+                requests=[
+                    GitlabProxy(
+                        GitlabProxyEnum.DATASET_V3,
+                        instance_id=instance_id,
+                    ),
+                ]
+            ).data
+        return self._cached_data

siibra/retrieval/requests.py

@@ -444,55 +444,6 @@ def get(self):
         return super().get()
 
 
-class EbrainsKgQuery(EbrainsRequest):
-    """Request outputs from a knowledge graph query."""
-
-    server = "https://kg.humanbrainproject.eu"
-    org = "minds"
-    domain = "core"
-    version = "v1.0.0"
-
-    SC_MESSAGES = {
-        401: "The provided EBRAINS authentication token is not valid",
-        403: "No permission to access the given query",
-        404: "Query with this id not found",
-    }
-
-    def __init__(self, query_id, instance_id=None, schema="dataset", params={}):
-        inst_tail = "/" + instance_id if instance_id is not None else ""
-        self.schema = schema
-        url = "{}/query/{}/{}/{}/{}/{}/instances{}?databaseScope=RELEASED".format(
-            self.server,
-            self.org,
-            self.domain,
-            self.schema,
-            self.version,
-            query_id,
-            inst_tail,
-        )
-        EbrainsRequest.__init__(
-            self,
-            url,
-            decoder=DECODERS[".json"],
-            params=params,
-            msg_if_not_cached=f"Executing EBRAINS KG query {query_id}{inst_tail}",
-        )
-
-    def get(self):
-        try:
-            result = EbrainsRequest.get(self)
-        except SiibraHttpRequestError as e:
-            if e.status_code in self.SC_MESSAGES:
-                raise RuntimeError(self.SC_MESSAGES[e.status_code])
-            else:
-                raise RuntimeError(
-                    f"Could not process HTTP request (status code: "
-                    f"{e.status_code}). Message was: {e.msg}"
-                    f"URL was: {e.url}"
-                )
-        return result
-
-
 def try_all_connectors():
     def outer(fn):
         @wraps(fn)