diff --git a/dsp_permissions_scripts/oap/oap_get.py b/dsp_permissions_scripts/oap/oap_get.py index 509a038c..239751c5 100644 --- a/dsp_permissions_scripts/oap/oap_get.py +++ b/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, diff --git a/dsp_permissions_scripts/oap/oap_set.py b/dsp_permissions_scripts/oap/oap_set.py index 5000cc79..868aef81 100644 --- a/dsp_permissions_scripts/oap/oap_set.py +++ b/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}")