resolve minhnh#11 (minhnh#12)

- add `types` arg to ModelBase.__init__, check whether `graph` or
  `types` is specified
- move type query logic to `get_node_types`
- update super classes and unit tests

src/rdf_utils/models/common.py

@@ -3,21 +3,39 @@
 from rdflib import URIRef, Graph, RDF
 
 
+def get_node_types(graph: Graph, node_id: URIRef) -> set[URIRef]:
+    """!
+    Get all types of a node in an RDF graph.
+
+    @param graph RDF graph to look up node types from
+    @param node_id URIRef of target node
+    @return set of the node's types as URIRef's
+    """
+    types = set()
+    for type_id in graph.objects(subject=node_id, predicate=RDF.type):
+        assert isinstance(type_id, URIRef), f"type '{type_id}' of node '{node_id}' not a URIRef"
+        types.add(type_id)
+    return types
+
+
 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:
+    def __init__(self, node_id: URIRef, graph: Optional[Graph] = None, types: Optional[set[URIRef]] = None) -> 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)
+        if graph is not None:
+            self.types = get_node_types(graph=graph, node_id=node_id)
+            assert types is None, f"ModelBase.__init__: node '{node_id}': both 'graph' and 'types' args are not None"
+        elif types is not None:
+            self.types = types
+        else:
+            raise RuntimeError(f"ModelBase.__init__: node '{node_id}': neither 'graph' or 'types' specified")
+        assert len(self.types) > 0, f"node '{self.id}' has no type"
 
-        assert len(self.types) > 0
         self._attributes = {}
 
     def has_attr(self, key: URIRef) -> bool:

src/rdf_utils/models/event_loop.py

@@ -20,8 +20,8 @@
 class EventReactionModel(ModelBase):
     event_id: URIRef
 
-    def __init__(self, graph: Graph, reaction_id: URIRef) -> None:
-        super().__init__(graph=graph, node_id=reaction_id)
+    def __init__(self, reaction_id: URIRef, graph: Graph) -> None:
+        super().__init__(node_id=reaction_id, graph=graph)
 
         evt_uri = graph.value(subject=self.id, predicate=URI_EL_PRED_REF_EVT)
         assert evt_uri is not None and isinstance(
@@ -33,8 +33,8 @@ def __init__(self, graph: Graph, reaction_id: URIRef) -> None:
 class FlagReactionModel(ModelBase):
     flag_id: URIRef
 
-    def __init__(self, graph: Graph, reaction_id: URIRef) -> None:
-        super().__init__(graph=graph, node_id=reaction_id)
+    def __init__(self, reaction_id: URIRef, graph: Graph) -> None:
+        super().__init__(node_id=reaction_id, graph=graph)
 
         flg_uri = graph.value(subject=self.id, predicate=URI_EL_PRED_REF_FLG)
         assert flg_uri is not None and isinstance(
@@ -49,8 +49,8 @@ class EventLoopModel(ModelBase):
     event_reactions: dict[URIRef, EventReactionModel]
     flag_reactions: dict[URIRef, FlagReactionModel]
 
-    def __init__(self, graph: Graph, el_id: URIRef) -> None:
-        super().__init__(graph=graph, node_id=el_id)
+    def __init__(self, el_id: URIRef, graph: Graph) -> None:
+        super().__init__(node_id=el_id, graph=graph)
 
         self.events_triggered = {}
         self.flag_values = {}
@@ -73,7 +73,7 @@ def __init__(self, graph: Graph, el_id: URIRef) -> None:
             assert isinstance(
                 evt_re_uri, URIRef
             ), f"EventReaction '{evt_re_uri}' is not of type URIRef: {type(evt_re_uri)}"
-            evt_re_model = EventReactionModel(graph=graph, reaction_id=evt_re_uri)
+            evt_re_model = EventReactionModel(reaction_id=evt_re_uri, graph=graph)
             assert (
                 evt_re_model.event_id in self.events_triggered
             ), f"'{evt_re_model.id}' reacts to event '{evt_re_model.event_id}', which is not in event loop '{self.id}'"
@@ -83,7 +83,7 @@ def __init__(self, graph: Graph, el_id: URIRef) -> None:
             assert isinstance(
                 flg_re_uri, URIRef
             ), f"FlagReaction '{flg_re_uri}' is not of type URIRef: {type(flg_re_uri)}"
-            flg_re_model = FlagReactionModel(graph=graph, reaction_id=flg_re_uri)
+            flg_re_model = FlagReactionModel(reaction_id=flg_re_uri, graph=graph)
             assert (
                 flg_re_model.flag_id in self.flag_values
             ), f"'{flg_re_model.id}' reacts to flag '{flg_re_model.flag_id}', which is not in event loop '{self.id}'"

tests/test_event_loop_model.py

@@ -109,19 +109,19 @@ def test_correct_el_model(self):
             check_shacl_constraints(graph=graph, shacl_dict={URL_MM_EL_SHACL: "turtle"})
         )
 
-        _ = EventLoopModel(graph=graph, el_id=URIREF_TEST_LOOP)
+        _ = EventLoopModel(el_id=URIREF_TEST_LOOP, graph=graph)
 
     def test_wrong_reactions(self):
         wrong_evt_g = Graph()
         wrong_evt_g.parse(data=EVT_LOOP_MODEL_WRONG_EVT, format="json-ld")
         with self.assertRaises(
             AssertionError, msg="not raised for reaction to an event not in loop"
         ):
-            _ = EventLoopModel(graph=wrong_evt_g, el_id=URIREF_TEST_LOOP)
+            _ = EventLoopModel(el_id=URIREF_TEST_LOOP, graph=wrong_evt_g)
         wrong_flg_g = Graph()
         wrong_flg_g.parse(data=EVT_LOOP_MODEL_WRONG_FLG, format="json-ld")
         with self.assertRaises(AssertionError, msg="not raised for reaction to a flag not in loop"):
-            _ = EventLoopModel(graph=wrong_flg_g, el_id=URIREF_TEST_LOOP)
+            _ = EventLoopModel(el_id=URIREF_TEST_LOOP, graph=wrong_flg_g)
 
 
 if __name__ == "__main__":

tests/test_python_model.py

@@ -57,7 +57,7 @@ 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))
+        os_model = ModelBase(node_id=URIRef(URI_OS_PATH_EXISTS), graph=graph)
         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))