refactor(xmlupload): change serialisation of values with rdflib (#1274)

src/dsp_tools/commands/xmlupload/models/serialise/serialise_rdf_value.py

@@ -0,0 +1,56 @@
+from abc import ABC
+from abc import abstractmethod
+from dataclasses import dataclass
+
+from rdflib import RDF
+from rdflib import BNode
+from rdflib import Graph
+from rdflib import Literal
+from rdflib import Namespace
+from rdflib import URIRef
+
+KNORA_API = Namespace("http://api.knora.org/ontology/knora-api/v2#")
+
+
+@dataclass(frozen=True)
+class ValueRDF(ABC):
+    resource_bn: BNode
+    prop_name: URIRef
+    value: Literal | URIRef
+    permissions: Literal | None
+    comment: Literal | None
+
+    @abstractmethod
+    def as_graph(self) -> Graph:
+        """Creates the value as rdflib graph"""
+
+    def _get_generic_graph(self, val_bn: BNode) -> Graph:
+        g = Graph()
+        g.add((self.resource_bn, self.prop_name, val_bn))
+        if self.permissions:
+            g.add((val_bn, KNORA_API.hasPermissions, self.permissions))
+        if self.comment:
+            g.add((val_bn, KNORA_API.valueHasComment, self.comment))
+        return g
+
+
+class BooleanValueRDF(ValueRDF):
+    value: Literal
+
+    def as_graph(self) -> Graph:
+        val_bn = BNode()
+        g = self._get_generic_graph(val_bn)
+        g.add((val_bn, RDF.type, KNORA_API.BooleanValue))
+        g.add((val_bn, KNORA_API.booleanValueAsBoolean, self.value))
+        return g
+
+
+class IntValueRDF(ValueRDF):
+    value: Literal
+
+    def as_graph(self) -> Graph:
+        val_bn = BNode()
+        g = self._get_generic_graph(val_bn)
+        g.add((val_bn, RDF.type, KNORA_API.IntValue))
+        g.add((val_bn, KNORA_API.intValueAsInt, self.value))
+        return g

src/dsp_tools/commands/xmlupload/resource_create_client.py

@@ -32,6 +32,8 @@
 from dsp_tools.commands.xmlupload.models.serialise.serialise_file_value import SerialiseMovingImageFileValue
 from dsp_tools.commands.xmlupload.models.serialise.serialise_file_value import SerialiseStillImageFileValue
 from dsp_tools.commands.xmlupload.models.serialise.serialise_file_value import SerialiseTextFileValue
+from dsp_tools.commands.xmlupload.models.serialise.serialise_rdf_value import BooleanValueRDF
+from dsp_tools.commands.xmlupload.models.serialise.serialise_rdf_value import IntValueRDF
 from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseColor
 from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseDecimal
 from dsp_tools.commands.xmlupload.models.serialise.serialise_value import SerialiseGeometry
@@ -339,43 +341,52 @@ def _make_boolean_prop(
 ) -> Graph:
     g = Graph()
     for value in prop.values:
-        single_val_bn = BNode()
-        g.add((res_bn, prop_name, single_val_bn))
-        g += _make_boolean_value(value, single_val_bn, permissions_lookup)
+        boolean_value = _make_boolean_value(value, prop_name, res_bn, permissions_lookup)
+        g += boolean_value.as_graph()
     return g
 
 
-def _make_boolean_value(value: XMLValue, val_bn: BNode, permissions_lookup: dict[str, Permissions]) -> Graph:
+def _make_boolean_value(
+    value: XMLValue, prop_name: URIRef, res_bn: BNode, permissions_lookup: dict[str, Permissions]
+) -> BooleanValueRDF:
     s = _assert_is_string(value.value)
-    g = Graph()
-    g.add((val_bn, RDF.type, KNORA_API.BooleanValue))
-    g.add((val_bn, KNORA_API.booleanValueAsBoolean, Literal(_to_boolean(s))))
-    _add_optional_permission_triple(value, val_bn, g, permissions_lookup)
-    if value.comment:
-        g.add((val_bn, KNORA_API.valueHasComment, Literal(value.comment)))
-    return g
+    as_bool = _to_boolean(s)
+    permission_literal = None
+    if permission_str := _get_permission_str(value.permissions, permissions_lookup):
+        permission_literal = Literal(permission_str)
+    return BooleanValueRDF(
+        resource_bn=res_bn,
+        prop_name=prop_name,
+        value=Literal(as_bool),
+        permissions=permission_literal,
+        comment=Literal(value.comment) if value.comment else None,
+    )
 
 
 def _make_integer_prop(
     prop: XMLProperty, res_bn: BNode, prop_name: URIRef, permissions_lookup: dict[str, Permissions]
 ) -> Graph:
     g = Graph()
     for value in prop.values:
-        single_val_bn = BNode()
-        g.add((res_bn, prop_name, single_val_bn))
-        g += _make_integer_value(value, single_val_bn, permissions_lookup)
+        int_value = _make_integer_value(value, prop_name, res_bn, permissions_lookup)
+        g += int_value.as_graph()
     return g
 
 
-def _make_integer_value(value: XMLValue, val_bn: BNode, permissions_lookup: dict[str, Permissions]) -> Graph:
+def _make_integer_value(
+    value: XMLValue, prop_name: URIRef, res_bn: BNode, permissions_lookup: dict[str, Permissions]
+) -> IntValueRDF:
     s = _assert_is_string(value.value)
-    g = Graph()
-    g.add((val_bn, RDF.type, KNORA_API.IntValue))
-    g.add((val_bn, KNORA_API.intValueAsInt, Literal(int(s))))
-    _add_optional_permission_triple(value, val_bn, g, permissions_lookup)
-    if value.comment:
-        g.add((val_bn, KNORA_API.valueHasComment, Literal(value.comment)))
-    return g
+    permission_literal = None
+    if permission_str := _get_permission_str(value.permissions, permissions_lookup):
+        permission_literal = Literal(permission_str)
+    return IntValueRDF(
+        resource_bn=res_bn,
+        prop_name=prop_name,
+        value=Literal(int(s)),
+        permissions=permission_literal,
+        comment=Literal(value.comment) if value.comment else None,
+    )
 
 
 def _make_interval_value(value: XMLValue) -> dict[str, Any]:

test/unittests/commands/xmlupload/test_resource_create_client.py

@@ -5,6 +5,7 @@
 from rdflib import RDF
 from rdflib import BNode
 from rdflib import Literal
+from rdflib import Namespace
 from rdflib import URIRef
 
 from dsp_tools.commands.xmlupload.models.deserialise.deserialise_value import IIIFUriInfo
@@ -22,6 +23,8 @@
 from dsp_tools.models.exceptions import BaseError
 from dsp_tools.models.exceptions import PermissionNotExistsError
 
+ONTO = Namespace("http://0.0.0.0:3333/ontology/9999/onto/v2#")
+
 
 class TestMakeBitstreamFileValue:
     """Tests the _make_bitstream_file_value function."""
@@ -460,7 +463,6 @@ def test_make_iiif_uri_value_serialised() -> None:
 
 def test_make_boolean_value_with_permissions() -> None:
     permissions_lookup = {"open": Permissions({PermissionValue.CR: ["knora-admin:ProjectAdmin"]})}
-
     xml_str = """
         <resource label="foo_1_label" restype=":foo_1_type" id="foo_1_id">
             <boolean-prop name=":isTrueOrFalse">
@@ -470,21 +472,20 @@ def test_make_boolean_value_with_permissions() -> None:
         """
     xmlresource = XMLResource.from_node(etree.fromstring(xml_str), "foo")
     test_val: XMLValue = xmlresource.properties[0].values[0]
-    b_node = BNode()
-    bool_graph = _make_boolean_value(test_val, b_node, permissions_lookup)
-
-    for triple in bool_graph.triples((b_node, None, None)):
-        if triple[1] == URIRef(RDF.type) and triple[2] == URIRef(KNORA_API.BooleanValue):
-            continue
-        elif triple[1] == URIRef(KNORA_API.booleanValueAsBoolean) and triple[2] == Literal(_to_boolean("true")):
-            continue
-        elif triple[1] == URIRef(KNORA_API.hasPermissions) and triple[2] == Literal(
-            str(permissions_lookup.get(str(test_val.permissions)))
-        ):
-            continue
-        else:
-            # unexpected triple
-            pytest.fail(f"unexpected triple: {triple}")
+    res_bn = BNode()
+    prop_name = ONTO.isTrueOrFalse
+    bool_graph = _make_boolean_value(
+        value=test_val, prop_name=prop_name, res_bn=res_bn, permissions_lookup=permissions_lookup
+    ).as_graph()
+    number_of_triples = 4
+    assert len(bool_graph) == number_of_triples
+    value_bn = next(bool_graph.objects(res_bn, prop_name))
+    rdf_type = next(bool_graph.objects(value_bn, RDF.type))
+    assert rdf_type == KNORA_API.BooleanValue
+    bool_val = next(bool_graph.objects(value_bn, KNORA_API.booleanValueAsBoolean))
+    assert bool_val == Literal(True)
+    permissions = next(bool_graph.objects(value_bn, KNORA_API.hasPermissions))
+    assert permissions == Literal(str(permissions_lookup.get("open")))
 
 
 if __name__ == "__main__":