From a006a5fbdfc9632adff18f0f14cb5269fb4beae0 Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Wed, 18 Sep 2024 13:36:24 +0200 Subject: [PATCH] move ModelBase and loader from bdd-dsl - move base class for models with ID and types from bdd-dsl - add ModelLoader and AttrLoaderProtocol to manage functions for loading attributes depending on application context - add attr loader function for Python module attributes - add import function that uses the loaded attributes --- src/rdf_utils/models.py | 51 ++++++++++++++++++++++++++++++++++++++ src/rdf_utils/python.py | 32 ++++++++++++++++++++++++ tests/test_python_model.py | 13 ++++++++-- 3 files changed, 94 insertions(+), 2 deletions(-) create mode 100644 src/rdf_utils/models.py diff --git a/src/rdf_utils/models.py b/src/rdf_utils/models.py new file mode 100644 index 0000000..b4e95b7 --- /dev/null +++ b/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) diff --git a/src/rdf_utils/python.py b/src/rdf_utils/python.py index d437368..ec11b9d 100644 --- a/src/rdf_utils/python.py +++ b/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) diff --git a/tests/test_python_model.py b/tests/test_python_model.py index 1157aec..381a8fc 100644 --- a/tests/test_python_model.py +++ b/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()