edit

dsp_permissions_scripts/oap/oap_get.py

@@ -1,5 +1,5 @@
 import warnings
-from typing import Iterable
+from typing import Any, Iterable
 from urllib.parse import quote_plus
 
 import requests
@@ -93,6 +93,39 @@ def _get_next_page(
     return False, []
 
 
+def get_resource(
+    resource_iri: str,
+    host: str,
+    token: str,
+) -> dict[str, Any]:
+    """Requests the resource with the given IRI from DSP-API"""
+    iri = quote_plus(resource_iri, safe="")
+    protocol = get_protocol(host)
+    url = f"{protocol}://{host}/v2/resources/{iri}"
+    headers = {"Authorization": f"Bearer {token}"}
+    response = http_call_with_retry(
+        action=lambda: requests.get(url, headers=headers, timeout=10),
+        err_msg=f"Error while getting resource {resource_iri}",
+    )
+    if response.status_code != 200:
+        raise ApiError( f"Error while getting resource {resource_iri}", response.text, response.status_code)
+    data: dict[str, Any] = response.json()
+    return data
+
+
+def get_oap_by_resource_iri(
+    host: str,
+    resource_iri: str,
+    token: str,
+) -> Oap:
+    resource = get_resource(
+        resource_iri=resource_iri,
+        host=host,
+        token=token,
+    )
+    scope = create_scope_from_string(resource["knora-api:hasPermissions"])
+    return Oap(scope=scope, object_iri=resource_iri)
+
 
 def get_all_resource_oaps_of_project(
     shortcode: str,

dsp_permissions_scripts/oap/oap_set.py

@@ -2,13 +2,13 @@
 
 import warnings
 from typing import Any
-from urllib.parse import quote_plus
 
 import requests
 
 from dsp_permissions_scripts.models.api_error import ApiError
 from dsp_permissions_scripts.models.scope import PermissionScope
 from dsp_permissions_scripts.models.value import ValueUpdate
+from dsp_permissions_scripts.oap.oap_get import get_resource
 from dsp_permissions_scripts.oap.oap_model import Oap
 from dsp_permissions_scripts.utils.authentication import get_protocol
 from dsp_permissions_scripts.utils.get_logger import get_logger
@@ -36,26 +36,6 @@ def _get_values_to_update(resource: dict[str, Any]) -> list[ValueUpdate]:
     return res
 
 
-def _get_resource(
-    resource_iri: str,
-    host: str,
-    token: str,
-) -> dict[str, Any]:
-    """Requests the resource with the given IRI from DSP-API"""
-    iri = quote_plus(resource_iri, safe="")
-    protocol = get_protocol(host)
-    url = f"{protocol}://{host}/v2/resources/{iri}"
-    headers = {"Authorization": f"Bearer {token}"}
-    response = http_call_with_retry(
-        action=lambda: requests.get(url, headers=headers, timeout=10),
-        err_msg=f"Error while getting resource {resource_iri}",
-    )
-    if response.status_code != 200:
-        raise ApiError( f"Error while getting resource {resource_iri}", response.text, response.status_code)
-    data: dict[str, Any] = response.json()
-    return data
-
-
 def _update_permissions_for_value(
     resource_iri: str,
     value: ValueUpdate,
@@ -142,7 +122,7 @@ def _update_permissions_for_resource_and_values(
 ) -> bool:
     """Updates the permissions for the given resource and its values on a DSP server"""
     try:
-        resource = _get_resource(resource_iri, host, token)
+        resource = get_resource(resource_iri, host, token)
     except Exception as exc:  # pylint: disable=broad-exception-caught
         logger.error(f"Cannot update resource {resource_iri}: {exc}")
         warnings.warn(f"Cannot update resource {resource_iri}: {exc}")