move ModelBase and loader from bdd-dsl

src/rdf_utils/models.py

@@ -0,0 +1,51 @@
+# SPDX-License-Identifier:  MPL-2.0
+from typing import Any, Dict, Optional, Protocol
+from rdflib import URIRef, Graph, RDF
+
+
+class ModelBase(object):
+    """All models should have an URI as ID and types"""
+
+    id: URIRef
+    types: set[URIRef]
+    _attributes: Dict[URIRef, Any]
+
+    def __init__(self, graph: Graph, node_id: URIRef) -> None:
+        self.id = node_id
+        self.types = set()
+        for type_id in graph.objects(subject=node_id, predicate=RDF.type):
+            assert isinstance(type_id, URIRef)
+            self.types.add(type_id)
+
+        assert len(self.types) > 0
+        self._attributes = {}
+
+    def has_attr(self, key: URIRef) -> bool:
+        return key in self._attributes
+
+    def set_attr(self, key: URIRef, val: Any) -> None:
+        self._attributes[key] = val
+
+    def get_attr(self, key: URIRef) -> Optional[Any]:
+        if key not in self._attributes:
+            return None
+
+        return self._attributes[key]
+
+
+class AttrLoaderProtocol(Protocol):
+    def __call__(self, graph: Graph, model: ModelBase, **kwargs: Any) -> None: ...
+
+
+class ModelLoader(object):
+    _loaders: list[AttrLoaderProtocol]
+
+    def __init__(self) -> None:
+        self._loaders = []
+
+    def register(self, loader: AttrLoaderProtocol) -> None:
+        self._loaders.append(loader)
+
+    def load_attributes(self, graph: Graph, model: ModelBase, **kwargs: Any):
+        for loader in self._loaders:
+            loader(graph=graph, model=model, **kwargs)

src/rdf_utils/python.py

@@ -3,6 +3,7 @@
 from typing import Any
 from rdflib import Graph, URIRef
 from rdf_utils.namespace import NS_MM_PYTHON
+from rdf_utils.models import ModelBase
 
 
 URI_PY_TYPE_MODULE_ATTR = NS_MM_PYTHON["ModuleAttribute"]
@@ -17,3 +18,34 @@ def import_attr_from_node(graph: Graph, uri: URIRef | str) -> Any:
     module_name = str(graph.value(uri, URI_PY_PRED_MODULE_NAME))
     attr_name = str(graph.value(uri, URI_PY_PRED_ATTR_NAME))
     return getattr(import_module(module_name), attr_name, None)
+
+
+def load_py_module_attr(graph: Graph, model: ModelBase, **kwargs: Any) -> None:
+    if URI_PY_TYPE_MODULE_ATTR not in model.types:
+        return
+
+    module_name = graph.value(model.id, URI_PY_PRED_MODULE_NAME)
+    assert (
+        module_name is not None
+    ), f"ModuleAttribute '{model.id}' doesn't have attr '{URI_PY_PRED_MODULE_NAME}'"
+    model.set_attr(key=URI_PY_PRED_MODULE_NAME, val=str(module_name))
+
+    attr_name = graph.value(model.id, URI_PY_PRED_ATTR_NAME)
+    assert (
+        attr_name is not None
+    ), f"ModuleAttribute '{model.id}' doesn't have attr '{URI_PY_PRED_ATTR_NAME}'"
+    model.set_attr(key=URI_PY_PRED_ATTR_NAME, val=str(attr_name))
+
+
+def import_attr_from_model(model: ModelBase) -> Any:
+    assert (
+        URI_PY_TYPE_MODULE_ATTR in model.types
+    ), f"model '{model.id}' doesn't have type '{URI_PY_TYPE_MODULE_ATTR}'"
+
+    module_name = model.get_attr(key=URI_PY_PRED_MODULE_NAME)
+    assert module_name is not None, f"module name not loaded for ModuleAttribute '{model.id}'"
+
+    attr_name = model.get_attr(key=URI_PY_PRED_ATTR_NAME)
+    assert attr_name is not None, f"attribute name not loaded for ModuleAttribute '{model.id}'"
+
+    return getattr(import_module(module_name), attr_name, None)

tests/test_python_model.py

@@ -2,10 +2,11 @@
 import unittest
 from urllib.request import urlopen
 import pyshacl
-from rdflib import ConjunctiveGraph
+from rdflib import ConjunctiveGraph, URIRef
+from rdf_utils.models import ModelBase, ModelLoader
 from rdf_utils.uri import URL_MM_PYTHON_JSON, URL_MM_PYTHON_SHACL, URL_SECORO_M
 from rdf_utils.resolver import install_resolver
-from rdf_utils.python import import_attr_from_node
+from rdf_utils.python import import_attr_from_model, import_attr_from_node, load_py_module_attr
 
 
 TEST_URL = f"{URL_SECORO_M}/models/tests"
@@ -31,6 +32,9 @@ def setUp(self):
         with urlopen(URL_MM_PYTHON_SHACL) as fp:
             self.mm_python_shacl_path = fp.file.name
 
+        self.model_loader = ModelLoader()
+        self.model_loader.register(load_py_module_attr)
+
     def test_python_import(self):
         graph = ConjunctiveGraph()
         graph.parse(data=PYTHON_MODEL, format="json-ld")
@@ -49,6 +53,11 @@ def test_python_import(self):
         os_path_exists = import_attr_from_node(graph, URI_OS_PATH_EXISTS)
         self.assertTrue(os_path_exists(self.mm_python_shacl_path))
 
+        os_model = ModelBase(graph=graph, node_id=URIRef(URI_OS_PATH_EXISTS))
+        self.model_loader.load_attributes(graph=graph, model=os_model)
+        os_path_exists = import_attr_from_model(os_model)
+        self.assertTrue(os_path_exists(self.mm_python_shacl_path))
+
 
 if __name__ == "__main__":
     unittest.main()