feat: support knora-base resources (DEV-3855)

dsp_permissions_scripts/oap/oap_get.py

@@ -1,4 +1,6 @@
+import copy
 from typing import Any
+from urllib.parse import quote
 from urllib.parse import quote_plus
 
 from dsp_permissions_scripts.models.errors import ApiError
@@ -30,6 +32,87 @@
     "knora-api:userHasPermission",
     "knora-api:hasPermissions",
 ]
+KB_RESCLASSES = [
+    "knora-api:VideoSegment",
+    "knora-api:AudioSegment",
+    "knora-api:Region",
+    "knora-api:Annotation",
+    "knora-api:LinkObj",
+]
+
+
+def get_oaps_of_kb_resclasses(dsp_client: DspClient, project_iri: str, oap_config: OapRetrieveConfig) -> list[Oap]:
+    match oap_config.retrieve_resources:
+        case "specified_res_classes":
+            specified_kb_resclasses = [x for x in KB_RESCLASSES if x in oap_config.specified_res_classes]
+            res_only_oaps = _get_oaps_of_specified_kb_resclasses(dsp_client, project_iri, specified_kb_resclasses)
+        case "all":
+            res_only_oaps = _get_oaps_of_specified_kb_resclasses(dsp_client, project_iri, KB_RESCLASSES)
+
+    match oap_config.retrieve_values:
+        case "none":
+            return res_only_oaps
+        case "specified_props":
+            enriched_oaps = _enrich_with_value_oaps(dsp_client, res_only_oaps, oap_config.specified_props)
+        case "all":
+            enriched_oaps = _enrich_with_value_oaps(dsp_client, res_only_oaps)
+
+    return enriched_oaps
+
+
+def _get_oaps_of_specified_kb_resclasses(
+    dsp_client: DspClient, project_iri: str, kb_resclasses: list[str]
+) -> list[Oap]:
+    logger.info(f"Retrieving OAPs from the knora-base resource classes {kb_resclasses}...")
+    all_oaps: list[Oap] = []
+    for resclass in kb_resclasses:
+        oaps = _get_oaps_of_one_kb_resclass(dsp_client, project_iri, resclass)
+        logger.info(f"Retrieved {len(oaps)} OAPs from class {resclass}.")
+        all_oaps.extend(oaps)
+    logger.info(f"Retrieved a total of {len(all_oaps)} OAPs from knora-base resource classes.")
+    return all_oaps
+
+
+def _enrich_with_value_oaps(
+    dsp_client: DspClient, res_only_oaps: list[Oap], restrict_to_props: list[str] | None = None
+) -> list[Oap]:
+    logger.info(f"Enriching {len(res_only_oaps)} OAPs of knora-base resources with their value OAPs...")
+    complete_oaps = copy.deepcopy(res_only_oaps)
+    for oap in complete_oaps:
+        full_resource = dsp_client.get(f"/v2/resources/{quote_plus(oap.resource_oap.resource_iri)}")
+        oap.value_oaps = _get_value_oaps(full_resource, restrict_to_props)
+    logger.info(f"Enriched {len(complete_oaps)} OAPs of knora-base resources with their value OAPs.")
+    return complete_oaps
+
+
+def _get_oaps_of_one_kb_resclass(dsp_client: DspClient, project_iri: str, resclass: str) -> list[Oap]:
+    oaps: list[Oap] = []
+    mayHaveMoreResults: bool = True
+    offset = 0
+    while mayHaveMoreResults:
+        sparql_query = """
+        PREFIX knora-api: <http://api.knora.org/ontology/knora-api/v2#>
+
+        CONSTRUCT {
+            ?kb_resclass knora-api:isMainResource true .
+        } WHERE {
+            BIND(<%(project_iri)s> as ?project_iri) .
+            ?kb_resclass a %(resclass)s .
+            ?kb_resclass knora-api:attachedToProject ?project_iri .
+        }
+        OFFSET %(offset)s
+        """ % {"resclass": resclass, "project_iri": project_iri, "offset": offset}  # noqa: UP031 (printf-string-formatting)
+        if not (response := dsp_client.get(f"/v2/searchextended/{quote(sparql_query, safe='')}")):
+            break  # if there are 0 results, the response is an empty dict
+        # 1 result: the resource is returned as a single dict
+        # >1 results: the resource is returned as a list of dicts
+        for json_resource in response.get("@graph", [response]):
+            scope = create_scope_from_string(json_resource["knora-api:hasPermissions"])
+            res_oap = ResourceOap(scope=scope, resource_iri=json_resource["@id"])
+            oaps.append(Oap(resource_oap=res_oap, value_oaps=[]))
+        mayHaveMoreResults = bool(response.get("knora-api:mayHaveMoreResults", False))
+        offset += 1
+    return oaps
 
 
 def _get_all_oaps_of_resclass(
@@ -156,5 +239,6 @@ def get_all_oaps_of_project(
     for resclass_localname in resclass_localnames:
         oaps = _get_all_oaps_of_resclass(resclass_localname, project_iri, dsp_client, oap_config)
         all_oaps.extend(oaps)
+    all_oaps.extend(get_oaps_of_kb_resclasses(dsp_client, project_iri, oap_config))
     logger.info(f"Retrieved a TOTAL of {len(all_oaps)} OAPs")
     return all_oaps