feat(xmlupload): check consistency with ontology on server (DEV-2921)

pyproject.toml

@@ -162,6 +162,7 @@ disable = [
     "no-else-raise",                 # would be less readable
     "logging-fstring-interpolation", # would be less readable
     "duplicate-code",                # too many false positives
+    "inconsistent-return-statements",# mypy can handle this better than pylint
     "invalid-name",                  # TODO: activate this
     "too-many-arguments",            # TODO: activate this
     "too-many-branches",             # TODO: activate this

src/dsp_tools/commands/xmlupload/check_consistency_with_ontology.py

@@ -0,0 +1,123 @@
+# sourcery skip: use-fstring-for-concatenation
+import regex
+from lxml import etree
+from regex import Pattern
+
+from dsp_tools.commands.xmlupload.models.ontology_diagnose_models import InvalidOntologyElements, OntoCheckInformation
+from dsp_tools.commands.xmlupload.ontology_client import OntologyClientLive
+from dsp_tools.models.exceptions import BaseError
+
+defaultOntologyColon: Pattern[str] = regex.compile(r"^:\w+$")
+knoraUndeclared: Pattern[str] = regex.compile(r"^\w+$")
+genericPrefixedOntology: Pattern[str] = regex.compile(r"^[A-Za-z]+-?[A-Za-z]+:\w+$")
+
+
+def do_xml_consistency_check(onto_client: OntologyClientLive, root: etree._Element) -> None:
+    """
+    This function takes an OntologyClient and the root of an XML.
+    It retrieves the ontologies from the server.
+    It iterates over the root.
+    If it finds any invalid properties or classes, they are printed out and a UserError is raised.
+
+     Args:
+         onto_client: client for the ontology retrieval
+         root: root of the XML
+
+     Raises:
+         UserError: if there are any invalid properties or classes
+    """
+    onto_check_info = OntoCheckInformation(
+        default_ontology_prefix=onto_client.default_ontology,
+        onto_lookup=onto_client.get_all_ontologies_from_server(),
+        save_location=onto_client.save_location,
+    )
+    classes, properties = _get_all_classes_and_properties(root)
+    _find_problems_in_classes_and_properties(classes, properties, onto_check_info)
+
+
+def _get_all_classes_and_properties(root: etree._Element) -> tuple[list[list[str]], list[list[str]]]:
+    classes = _get_all_class_types_and_ids(root)
+    properties = []
+    for resource in root.iterchildren(tag="resource"):
+        properties.extend(_get_all_property_names_and_resource_ids(resource))
+    return classes, properties
+
+
+def _get_all_class_types_and_ids(root: etree._Element) -> list[list[str]]:
+    return [[resource.attrib["id"], resource.attrib["restype"]] for resource in root.iterchildren(tag="resource")]
+
+
+def _get_all_property_names_and_resource_ids(resource: etree._Element) -> list[list[str]]:
+    return [[resource.attrib["id"], prop.attrib["name"]] for prop in resource.iterchildren() if prop.tag != "bitstream"]
+
+
+def _find_problems_in_classes_and_properties(
+    classes: list[list[str]], properties: list[list[str]], onto_check_info: OntoCheckInformation
+) -> None:
+    class_problems = _diagnose_all_classes(classes, onto_check_info)
+    property_problems = _diagnose_all_properties(properties, onto_check_info)
+    if not class_problems and not property_problems:
+        return None
+    problems = InvalidOntologyElements(
+        save_path=onto_check_info.save_location, classes=class_problems, properties=property_problems
+    )
+    problems.execute_problem_protocol()
+
+
+def _diagnose_all_classes(classes: list[list[str]], onto_check_info: OntoCheckInformation) -> list[list[str]]:
+    problem_list = []
+    for cls_info in classes:
+        if problem := _diagnose_class(cls_info, onto_check_info):
+            problem_list.append(problem)
+    return problem_list
+
+
+def _diagnose_class(class_info: list[str], onto_check_info: OntoCheckInformation) -> list[str] | None:
+    try:
+        prefix, cls_ = _get_prefix_and_prop_cls_identifier(class_info[1], onto_check_info.default_ontology_prefix)
+    except BaseError:
+        class_info.append("Resource type does not follow a known ontology pattern")
+        return class_info
+    try:
+        if cls_ not in onto_check_info.onto_lookup[prefix].classes:
+            class_info.append("Invalid Class Type")
+            return class_info
+        return None
+    except KeyError:
+        class_info.append("Unknown ontology prefix")
+        return class_info
+
+
+def _diagnose_all_properties(properties: list[list[str]], onto_check_info: OntoCheckInformation) -> list[list[str]]:
+    problem_list = []
+    for prop_info in properties:
+        if problem := _diagnose_properties(prop_info, onto_check_info):
+            problem_list.append(problem)
+    return problem_list
+
+
+def _diagnose_properties(prop_info: list[str], onto_check_info: OntoCheckInformation) -> list[str] | None:
+    try:
+        prefix, prop = _get_prefix_and_prop_cls_identifier(prop_info[1], onto_check_info.default_ontology_prefix)
+    except BaseError:
+        prop_info.append("Property name does not follow a known ontology pattern")
+        return prop_info
+    try:
+        if prop not in onto_check_info.onto_lookup[prefix].properties:
+            prop_info.append("Invalid Property")
+            return prop_info
+        return None
+    except KeyError:
+        prop_info.append("Unknown ontology prefix")
+        return prop_info
+
+
+def _get_prefix_and_prop_cls_identifier(prop_cls: str, default_ontology_prefix: str) -> tuple[str, ...]:
+    if defaultOntologyColon.match(prop_cls):
+        return default_ontology_prefix, prop_cls.lstrip(":")
+    elif knoraUndeclared.match(prop_cls):
+        return "knora-api", prop_cls
+    elif genericPrefixedOntology.match(prop_cls):
+        return tuple(prop_cls.split(":"))
+    else:
+        raise BaseError(f"The input property or class: '{prop_cls}' does not follow a known ontology pattern.")

src/dsp_tools/commands/xmlupload/models/ontology_diagnose_models.py

@@ -0,0 +1,102 @@
+# sourcery skip: use-fstring-for-concatenation
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import pandas as pd
+
+from dsp_tools.models.exceptions import UserError
+
+
+@dataclass(frozen=True)
+class Ontology:
+    """This class saves the properties and the classes from an ontology."""
+
+    classes: list[str] = field(default_factory=list)
+    properties: list[str] = field(default_factory=list)
+
+
+@dataclass
+class OntoCheckInformation:
+    """This class saves information that is needed to check if consistency with the ontology."""
+
+    default_ontology_prefix: str
+    onto_lookup: dict[str, Ontology]
+    save_location: Path
+
+
+@dataclass(frozen=True)
+class InvalidOntologyElements:
+    """This class saves and prints out the information regarding ontology classes and properties
+    that are in the XML but not the ontology."""
+
+    save_path: Path
+    classes: list[list[str]] = field(default_factory=list)
+    properties: list[list[str]] = field(default_factory=list)
+
+    def execute_problem_protocol(self) -> None:
+        """
+        This function is executed if there are any elements in properties or classes.
+        If there are more than 100 entries combined, then the result is also saved as an excel.
+
+        Raises:
+            UserError: If properties or classes have any entries.
+        """
+        extra_separator = "\n\n----------------------------\n\n"
+        msg = "Some property and or class type(s) used in the XML are unknown:" + extra_separator
+        cls_msg = self._print_problem_string_cls()
+        if cls_msg:
+            msg += cls_msg + extra_separator
+        prop_msg = self._print_problem_string_props()
+        if prop_msg:
+            msg += prop_msg
+        if len(self.classes) + len(self.properties) > 50:
+            ex_name = "InvalidOntologyElements_in_XML.xlsx"
+            self._get_and_save_problems_as_excel(ex_name)
+            msg += extra_separator + f"\nAn excel: '{ex_name}' was saved at '{self.save_path}' listing the problems."
+        raise UserError(msg)
+
+    def _get_and_save_problems_as_excel(self, excel_name: str) -> None:
+        df = self._get_problems_as_df()
+        df.to_excel(excel_writer=Path(self.save_path, excel_name), sheet_name=" ", index=False)
+
+    def _get_problems_as_df(self) -> pd.DataFrame:
+        problems = [
+            {
+                "resource id": x[0],
+                "problematic type": x[1],
+                "problem": x[2],
+            }
+            for x in self.classes
+        ]
+        problems.extend(
+            [
+                {
+                    "resource id": x[0],
+                    "problematic type": x[1],
+                    "problem": x[2],
+                }
+                for x in self.properties
+            ]
+        )
+        return pd.DataFrame.from_records(problems)
+
+    def _print_problem_string_cls(self) -> str:
+        if self.classes:
+            separator = "\n----------------------------\n"
+            problems = [
+                f"\tResource ID: '{x[0]}'\n\tResource Type: '{x[1]}'\n\tProblem: '{x[2]}'" for x in self.classes
+            ]
+            return "The following resource(s) have an invalid resource type:\n\n" + separator.join(problems)
+        else:
+            return ""
+
+    def _print_problem_string_props(self) -> str:
+        if self.properties:
+            separator = "\n----------------------------\n"
+            problems = [
+                f"\tResource ID: '{x[0]}'\n\tProperty Name: '{x[1]}'\n\tProblem: '{x[2]}'" for x in self.properties
+            ]
+            return "The following resource(s) have invalid property type(s):\n\n" + separator.join(problems)
+        else:
+            return ""

src/dsp_tools/commands/xmlupload/ontology_client.py

@@ -0,0 +1,133 @@
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Protocol
+
+from dsp_tools.commands.xmlupload.models.ontology_diagnose_models import Ontology
+from dsp_tools.models.exceptions import BaseError, UserError
+from dsp_tools.utils.connection import Connection
+from dsp_tools.utils.create_logger import get_logger
+from dsp_tools.utils.shared import try_network_action
+
+# pylint: disable=too-few-public-methods
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class OntologyClient(Protocol):
+    """Interface (protocol) for ontology-related requests to the DSP-API."""
+
+    con: Connection
+    shortcode: str
+    default_ontology: str
+    save_location: Path
+    ontology_names: list[str] = field(default_factory=list)
+
+    def get_all_ontologies_from_server(self) -> dict[str, Ontology]:
+        """Get all the ontologies for a project and the knora-api ontology from the server."""
+
+
+@dataclass
+class OntologyClientLive:
+    """Client handling ontology-related requests to the DSP-API."""
+
+    con: Connection
+    shortcode: str
+    default_ontology: str
+    save_location: Path
+    ontology_names: list[str] = field(default_factory=list)
+
+    def get_all_ontologies_from_server(self) -> dict[str, Ontology]:
+        """
+        This function returns all the project and knora-api ontology that are on the server.
+
+        Returns:
+            a dictionary with the ontology name as key and the ontology as value.
+
+        """
+        ontologies = self._get_all_ontology_jsons_from_server()
+        return {onto_name: deserialize_ontology(onto_graph) for onto_name, onto_graph in ontologies.items()}
+
+    def _get_all_ontology_jsons_from_server(self) -> dict[str, list[dict[str, Any]]]:
+        self._get_ontology_names_from_server()
+        project_ontos = {onto: self._get_ontology_from_server(onto) for onto in self.ontology_names}
+        project_ontos["knora-api"] = self._get_knora_api_from_server()
+        return project_ontos
+
+    def _get_ontology_names_from_server(self) -> None:
+        try:
+            url = f"/admin/projects/shortcode/{self.shortcode}"
+            res: dict[str, Any] = try_network_action(self.con.get, route=url)
+        except BaseError as e:
+            raise UserError(f"A project with shortcode {self.shortcode} could not be found on the DSP server") from e
+        try:
+            onto_iris: list[str] = res["project"]["ontologies"]
+        except KeyError as e:
+            raise BaseError(f"Unexpected response from server: {res}") from e
+        onto_names: list[str] = [iri.split("/")[-1] for iri in onto_iris]
+        self.ontology_names = onto_names
+
+    def _get_ontology_from_server(self, ontology_name: str) -> list[dict[str, Any]]:
+        try:
+            url = f"/ontology/{self.shortcode}/{ontology_name}/v2"
+            res: dict[str, Any] = try_network_action(self.con.get, route=url)
+        except BaseError:
+            raise BaseError(
+                f"Ontologies for project {self.shortcode} could not be retrieved from the DSP server"
+            ) from None
+        try:
+            onto_graph: list[dict[str, Any]] = res["@graph"]
+        except KeyError as e:
+            raise BaseError(f"Unexpected response from server: {res}") from e
+        return onto_graph
+
+    def _get_knora_api_from_server(self) -> list[dict[str, Any]]:
+        url = "/ontology/knora-api/v2#"
+        try:
+            res: dict[str, Any] = try_network_action(self.con.get, route=url)
+        except BaseError:
+            raise BaseError("Knora-api ontology could not be retrieved from the DSP server") from None
+        try:
+            onto_graph: list[dict[str, Any]] = res["@graph"]
+        except KeyError as e:
+            raise BaseError(f"Unexpected response from server when retrieving knora-api ontology: {res}") from e
+        return onto_graph
+
+
+def deserialize_ontology(onto_graph: list[dict[str, Any]]) -> Ontology:
+    """
+    This function takes an ontology graph from the DSP-API.
+    It extracts the classes and properties.
+    And saves them in an instance of the class Ontology.
+
+    Args:
+        onto_graph: graph from DSP-API
+
+    Returns:
+        Ontology instance with the classes and properties
+    """
+    classes = _get_all_cleaned_classes_from_graph(onto_graph)
+    properties = _get_all_cleaned_properties_from_graph(onto_graph)
+    return Ontology(classes, properties)
+
+
+def _get_all_cleaned_classes_from_graph(onto_graph: list[dict[str, Any]]) -> list[str]:
+    classes = _get_all_classes_from_graph(onto_graph)
+    return _remove_prefixes(classes)
+
+
+def _get_all_classes_from_graph(onto_graph: list[dict[str, Any]]) -> list[str]:
+    return [elem["@id"] for elem in onto_graph if elem.get("knora-api:isResourceClass")]
+
+
+def _get_all_cleaned_properties_from_graph(onto_graph: list[dict[str, Any]]) -> list[str]:
+    props = _get_all_properties_from_graph(onto_graph)
+    return _remove_prefixes(props)
+
+
+def _get_all_properties_from_graph(onto_graph: list[dict[str, Any]]) -> list[str]:
+    return [elem["@id"] for elem in onto_graph if not elem.get("knora-api:isResourceClass")]
+
+
+def _remove_prefixes(ontology_elements: list[str]) -> list[str]:
+    return [x.split(":")[1] for x in ontology_elements]

src/dsp_tools/commands/xmlupload/xmlupload.py

@@ -10,12 +10,14 @@
 
 from lxml import etree
 
+from dsp_tools.commands.xmlupload.check_consistency_with_ontology import do_xml_consistency_check
 from dsp_tools.commands.xmlupload.iri_resolver import IriResolver
 from dsp_tools.commands.xmlupload.list_client import ListClient, ListClientLive
 from dsp_tools.commands.xmlupload.models.permission import Permissions
 from dsp_tools.commands.xmlupload.models.sipi import Sipi
 from dsp_tools.commands.xmlupload.models.xmlpermission import XmlPermission
 from dsp_tools.commands.xmlupload.models.xmlresource import BitstreamInfo, XMLResource
+from dsp_tools.commands.xmlupload.ontology_client import OntologyClientLive
 from dsp_tools.commands.xmlupload.project_client import ProjectClient, ProjectClientLive
 from dsp_tools.commands.xmlupload.read_validate_xml_file import validate_and_parse_xml_file
 from dsp_tools.commands.xmlupload.resource_create_client import ResourceCreateClient
@@ -84,6 +86,11 @@ def xmlupload(
     con = login(server=server, user=user, password=password, dump=config.diagnostics.dump)
     sipi_server = Sipi(sipi, con.get_token())
 
+    ontology_client = OntologyClientLive(
+        con=con, shortcode=shortcode, default_ontology=default_ontology, save_location=config.diagnostics.save_location
+    )
+    do_xml_consistency_check(onto_client=ontology_client, root=root)
+
     resources, permissions_lookup, stash = _prepare_upload(
         root=root,
         con=con,