BlueBrainNexusStore: fix tagging schemas (BlueBrain#327)

* BlueBrainNexusStore: tagging schemas required a specific schema URI to be present in the request URL or '_' in case of UNCONSTRAINED resources

* Updated _tag_many action error message formatting to match _tag_one error message formatting

kgforge/specializations/stores/bluebrain_nexus.py

@@ -64,7 +64,7 @@
 from kgforge.core.wrappings.paths import Filter, create_filters_from_dict
 from kgforge.specializations.mappers import DictionaryMapper
 from kgforge.specializations.mappings import DictionaryMapping
-from kgforge.specializations.stores.nexus.service import BatchAction, Service
+from kgforge.specializations.stores.nexus.service import BatchAction, Service, _error_message
 
 
 class CategoryDataType(Enum):
@@ -620,10 +620,10 @@ def _tag_many(self, resources: List[Resource], value: str) -> None:
         )
 
     def _tag_one(self, resource: Resource, value: str) -> None:
-        data = {"tag": value, "rev": resource._store_metadata._rev}
-        url = f"{self.service.url_resources}/_/{quote_plus(resource.id)}/tags?rev={resource._store_metadata._rev}"
+        url, data, rev_param = self.service._prepare_tag(resource, value)
         try:
-            params_tag = copy.deepcopy(self.service.params.get("tag", None))
+            params_tag = copy.deepcopy(self.service.params.get("tag", {}))
+            params_tag.update(rev_param)
             response = requests.post(
                 url,
                 headers=self.service.headers,
@@ -1071,29 +1071,6 @@ def rewrite_uri(self, uri: str, context: Context, **kwargs) -> str:
         return uri
 
 
-def _error_message(error: HTTPError) -> str:
-    def format_message(msg):
-        return "".join([msg[0].lower(), msg[1:-1], msg[-1] if msg[-1] != "." else ""])
-
-    try:
-        error_json = error.response.json()
-        messages = []
-        reason = error_json.get("reason", None)
-        details = error_json.get("details", None)
-        if reason:
-            messages.append(format_message(reason))
-        if details:
-            messages.append(format_message(details))
-        messages = messages if reason or details else [str(error)]
-        return ". ".join(messages)
-    except Exception as e:
-        pass
-    try:
-        return format_message(error.response.text())
-    except Exception:
-        return format_message(str(error))
-
-
 def _create_select_query(vars_, statements, distinct, search_in_graph):
     where_clauses = (
         f"{{ Graph ?g {{{statements}}}}}" if search_in_graph else f"{{{statements}}}"

kgforge/specializations/stores/demo_store.py

@@ -239,7 +239,7 @@ def _archive_id(rid: str, version: int) -> str:
     def _tag_id(rid: str, tag: str) -> str:
         return f"{rid}_tag={tag}"
 
-    def rewrite_ur(self, uri: str, context: Context, **kwargs) -> str:
+    def rewrite_uri(self, uri: str, context: Context, **kwargs) -> str:
         raise not_supported()
 
     class RecordExists(Exception):

kgforge/specializations/stores/nexus/service.py

@@ -74,6 +74,7 @@ class Service:
     DEFAULT_ES_INDEX_FALLBACK = f"{NEXUS_NAMESPACE_FALLBACK}defaultElasticSearchIndex"
     DEPRECATED_PROPERTY_FALLBACK = f"{NEXUS_NAMESPACE_FALLBACK}deprecated"
     PROJECT_PROPERTY_FALLBACK = f"{NEXUS_NAMESPACE_FALLBACK}project"
+    UNCONSTRAINED_SCHEMA = "https://bluebrain.github.io/nexus/schemas/unconstrained.json"
 
     def __init__(
         self,
@@ -278,7 +279,7 @@ def resolve_context(self, iri: str, local_only: Optional[bool] = False) -> Dict:
                 document.remove(self.store_local_context)
         self.context_cache.update({context_to_resolve: document})
         return document
-
+    
     def batch_request(
         self,
         resources: List[Resource],
@@ -347,12 +348,8 @@ def create_tasks(
                     )
 
                 if batch_action == batch_action.TAG:
-                    url = "/".join(
-                        (self.url_resources, "_", quote_plus(resource.id), "tags")
-                    )
-                    rev = resource._store_metadata._rev
-                    params["rev"] = rev
-                    payload = {"tag": kwargs.get("tag"), "rev": rev}
+                    url, payload, rev_param = self._prepare_tag(resource, kwargs.get("tag"))
+                    params.update(rev_param)
                     prepared_request = loop.create_task(
                         queue(
                             hdrs.METH_POST,
@@ -434,8 +431,7 @@ async def request(method, session, url, resource, payload, params, exception):
                 if response.status < 400:
                     return BatchResult(resource, content)
                 else:
-                    msg = " ".join(re.findall("[A-Z][^A-Z]*", content["@type"])).lower()
-                    error = exception(msg)
+                    error = exception(_error_message(content))
                     return BatchResult(resource, error)
 
         async def dispatch_action():
@@ -449,6 +445,15 @@ async def dispatch_action():
 
         return asyncio.run(dispatch_action())
 
+    def _prepare_tag(self, resource, tag) -> Tuple[str, str, str]:
+        schema_id = resource._store_metadata._constrainedBy
+        schema_id = "_" if schema_id == self.UNCONSTRAINED_SCHEMA or schema_id is None else schema_id
+        url = "/".join((self.url_resources, quote_plus(schema_id), quote_plus(resource.id), "tags"))
+        rev = resource._store_metadata._rev
+        params = {"rev":rev}
+        data = {"tag": tag, "rev": rev}
+        return url, data, params
+
     def sync_metadata(self, resource: Resource, result: Dict) -> None:
         metadata = (
             {"id": resource.id}
@@ -582,3 +587,27 @@ def to_resource(
         if not hasattr(resource, "id") and kwargs and 'id' in kwargs.keys():
             resource.id = kwargs.get("id")
         return resource
+
+
+def _error_message(error: Union[HTTPError, Dict]) -> str:
+    def format_message(msg):
+        return "".join([msg[0].lower(), msg[1:-1], msg[-1] if msg[-1] != "." else ""])
+
+    try:
+        error_json = error.response.json() if isinstance(error, HTTPError) else error
+        messages = []
+        reason = error_json.get("reason", None)
+        details = error_json.get("details", None)
+        if reason:
+            messages.append(format_message(reason))
+        if details:
+            messages.append(format_message(details))
+        messages = messages if reason or details else [str(error)]
+        return ". ".join(messages)
+    except Exception as e:
+        pass
+    try:
+        error_text =  error.response.text() if isinstance(error, HTTPError) else str(error)
+        return format_message(error_text)
+    except Exception:
+        return format_message(str(error))

tests/specializations/stores/test_bluebrain_nexus.py

@@ -11,18 +11,16 @@
 #
 # You should have received a copy of the GNU Lesser General Public License
 # along with Blue Brain Nexus Forge. If not, see <https://choosealicense.com/licenses/lgpl-3.0/>.
-import json
 
 import os
 from unittest import mock
-from urllib.parse import urljoin
+from urllib.parse import quote_plus, urljoin
 from urllib.request import pathname2url
 from uuid import uuid4
 
 import nexussdk
 import pytest
 from typing import Callable, Union, List
-from collections import OrderedDict
 
 from kgforge.core import Resource
 from kgforge.core.archetypes import Store
@@ -76,6 +74,7 @@ def metadata_data_compacted():
         "_deprecated": False,
         "_updatedBy": "http://integration.kfgorge.test",
         "_rev": 1,
+        "_constrainedBy":"http://schema.org/Building"
     }
 
 
@@ -98,6 +97,7 @@ def registered_building(building, model_context, store_metadata_value):
     else:
         building.id = f"{urljoin('file:', pathname2url(os.getcwd()))}/{str(uuid4())}"
     store_metadata_value["id"] = building.id
+    store_metadata_value["_constrainedBy"] = "http://schema.org/Building"
     building._store_metadata = wrap_dict(store_metadata_value)
     return building
 
@@ -176,6 +176,33 @@ def test_to_resource(nexus_store, registered_building, building_jsonld):
     assert str(result._store_metadata) == str(registered_building._store_metadata)
 
 
+@pytest.mark.parametrize("schema, expected_params, expected_url_template",
+                         [
+                            pytest.param(       
+                                ("http://schema.org/Building"),
+                                ({"rev":1}),
+                                ("/".join((NEXUS,"resources",BUCKET, quote_plus("http://schema.org/Building"),"{}", "tags"))),
+                                id="tag-connstrained",
+                            ),
+                            pytest.param(       
+                                (Service.UNCONSTRAINED_SCHEMA),
+                                ({"rev":1}),
+                                ("/".join((NEXUS,"resources",BUCKET, quote_plus("_"),"{}", "tags"))),
+                                id="tag-unconstrained",
+                            )
+                         ])
+def test_prepare_tag(nexus_store, registered_building, schema, expected_params, expected_url_template):
+    tagValue = "aTag"
+    registered_building._store_metadata._constrainedBy = schema
+    url, data, params = nexus_store.service._prepare_tag(registered_building, tagValue)
+    expected_data = {"tag":tagValue, "rev":registered_building._store_metadata._rev}
+    expected_url = expected_url_template.format(quote_plus(registered_building.id))
+
+    assert params == expected_params
+    assert data == expected_data
+    assert url == expected_url
+
+
 @pytest.mark.parametrize("url,is_file, expected",
                          [
                             pytest.param(