reorganise init of rdf model service

kgforge/specializations/models/rdf/rdf_model_service.py

@@ -12,16 +12,10 @@
 # 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 types
-from io import StringIO
+from abc import abstractmethod, ABC
 from typing import List, Dict, Tuple, Set, Optional
-from abc import abstractmethod
-from typing import List, Dict, Tuple, Set, Optional
-from pyshacl.shape import Shape
-from pyshacl.shapes_graph import ShapesGraph
-from rdflib import Graph, URIRef, RDF, XSD
-from rdflib.plugins.sparql.results.jsonresults import JSONResultSerializer
 
+from rdflib import Graph, URIRef, RDF, XSD
 from kgforge.core.commons.sparql_query_builder import SPARQLQueryBuilder
 from kgforge.core.resource import Resource
 from kgforge.core.commons.context import Context
@@ -30,23 +24,21 @@
 
 from kgforge.specializations.models.rdf.node_properties import NodeProperties
 from kgforge.specializations.models.rdf.utils import as_term
+from kgforge.specializations.models.rdf.pyshacl_shape_wrapper import ShapesGraphWrapper
 
 
-class RdfModelService:
+class RdfModelService(ABC):
 
-    def __init__(
-            self, graph: Graph,
-            shape_to_source: Dict[URIRef, str],
-            class_to_shape:  Dict[str, URIRef],
-            context_iri: Optional[str] = None,
-    ) -> None:
+    def __init__(self, context_iri: Optional[str] = None):
 
         if context_iri is None:
             raise ConfigurationError("RdfModel requires a context")
-        self._graph = graph
+
+        self._graph, self.shape_to_source, self.class_to_shape =  self._build_shapes_map()
+        self._shapes_graph = ShapesGraphWrapper(self._graph)
+
         self._context_cache = dict()
-        self.shape_to_source = shape_to_source
-        self.class_to_shape = class_to_shape
+
         self.context = Context(self.resolve_context(context_iri), context_iri)
         self.types_to_shapes: Dict[str, URIRef] = self._build_types_to_shapes()
 
@@ -92,17 +84,21 @@ def validate(self, resource: Resource, type_: str):
 
     @abstractmethod
     def _validate(self, iri: str, data_graph: Graph) -> Tuple[bool, Graph, str]:
-        raise NotImplementedError()
+        ...
 
     @abstractmethod
     def resolve_context(self, iri: str) -> Dict:
         """For a given IRI return its resolved context recursively"""
-        raise NotImplementedError()
+        ...
 
     @abstractmethod
     def generate_context(self) -> Dict:
         """Generates a JSON-LD context with the classes and terms present in the SHACL graph."""
-        raise NotImplementedError()
+        ...
+
+    @abstractmethod
+    def _build_shapes_map(self) -> Tuple[Graph, Dict[URIRef, str], Dict[str, URIRef]]:
+        ...
 
     def _build_types_to_shapes(self) -> Dict[str, URIRef]:
         """Iterates the classes_to_shapes dictionary to create a term to shape dictionary filtering

kgforge/specializations/models/rdf/rdf_model_service_from_directory.py

@@ -13,7 +13,7 @@
 # along with Blue Brain Nexus Forge. If not, see <https://choosealicense.com/licenses/lgpl-3.0/>.
 import os
 from pathlib import Path
-from typing import Dict, Tuple, Optional
+from typing import Dict, Tuple
 
 from pyshacl import validate
 from rdflib import Graph, URIRef
@@ -22,20 +22,13 @@
 from kgforge.core.commons.context import Context
 from kgforge.specializations.models.rdf.node_properties import NodeProperties
 from kgforge.specializations.models.rdf.rdf_model_service import RdfModelService
-from kgforge.specializations.models.rdf.pyshacl_shape_wrapper import ShapesGraphWrapper
 
 
 class RdfModelServiceFromDirectory(RdfModelService):
 
     def __init__(self, dir_path: Path, context_iri: str) -> None:
-
-        graph, shape_to_source, class_to_shape = self._build_shapes_map(dir_path=dir_path)
-        self._shapes_graph = ShapesGraphWrapper(graph)
-
-        super().__init__(
-            graph=graph, context_iri=context_iri, shape_to_source=shape_to_source,
-            class_to_shape=class_to_shape
-        )
+        self.dir_path = dir_path
+        super().__init__(context_iri=context_iri)
 
     def materialize(self, iri: URIRef) -> NodeProperties:
         sh = self._shapes_graph.lookup_shape_from_node(iri)
@@ -55,17 +48,15 @@ def resolve_context(self, iri: str) -> Dict:
         try:
             context = Context(iri)
         except FileNotFoundError as e:
-            raise ValueError(e)
+            raise ValueError(e) from e
 
         self._context_cache.update({iri: context.document})
         return context.document
 
     def generate_context(self) -> Dict:
         return self._generate_context()
 
-    def _build_shapes_map(
-            self, dir_path: Path
-    ) -> Tuple[Graph, Dict[URIRef, str], Dict[str, URIRef]]:
+    def _build_shapes_map(self) -> Tuple[Graph, Dict[URIRef, str], Dict[str, URIRef]]:
 
         query = """
                     PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
@@ -86,7 +77,7 @@ def _build_shapes_map(
         graph = Graph()
 
         extensions = [".ttl", ".n3", ".json", ".rdf"]
-        for f in dir_path.rglob(os.path.join("*.*")):
+        for f in self.dir_path.rglob(os.path.join("*.*")):
             graph_i = Graph()
             if f.suffix in extensions:
                 file_format = guess_format(f.name)

kgforge/specializations/models/rdf/rdf_model_service_from_store.py

@@ -46,13 +46,7 @@ def __init__(self, default_store: Store, context_iri: Optional[str] = None,
 
         self._imported = []
 
-        graph, shape_to_resource, class_to_shape = self._build_shapes_map()
-        self._shapes_graph = ShapesGraphWrapper(graph)
-
-        super().__init__(
-            graph=graph, context_iri=context_iri, shape_to_source=shape_to_resource,
-            class_to_shape=class_to_shape
-        )
+        super().__init__(context_iri=context_iri)
 
     def materialize(self, iri: URIRef) -> NodeProperties:
         shape: ShapeWrapper = self._load_and_get_type_shape(iri)