Merge branch 'main' into wip/handle-duplicates

dsp_permissions_scripts/ap/ap_delete.py

@@ -1,53 +1,35 @@
-import warnings
 from urllib.parse import quote_plus
 
-import requests
-
 from dsp_permissions_scripts.ap.ap_model import Ap
-from dsp_permissions_scripts.models.api_error import ApiError
-from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.models.errors import ApiError
+from dsp_permissions_scripts.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
-from dsp_permissions_scripts.utils.try_request import http_call_with_retry
 
 logger = get_logger(__name__)
 
 
-def _delete_ap_on_server(
-    ap: Ap,
-    host: str,
-    token: str,
-) -> None:
-    headers = {"Authorization": f"Bearer {token}"}
+def _delete_ap_on_server(ap: Ap, dsp_client: DspClient) -> None:
     ap_iri = quote_plus(ap.iri, safe="")
-    protocol = get_protocol(host)
-    url = f"{protocol}://{host}/admin/permissions/{ap_iri}"
-    response = http_call_with_retry(
-        action=lambda: requests.delete(url, headers=headers, timeout=20),
-        err_msg=f"Could not delete Administrative Permission {ap.iri}",
-    )
-    if response.status_code != 200:
-        raise ApiError(f"Could not delete Administrative Permission {ap.iri}", response.text, response.status_code)
+    try:
+        dsp_client.delete(f"/admin/permissions/{ap_iri}")
+    except ApiError as err:
+        err.message = f"Could not delete Administrative Permission {ap.iri}"
+        raise err from None
 
 
 def delete_ap_of_group_on_server(
     host: str,
-    token: str,
     existing_aps: list[Ap],
     forGroup: str,
+    dsp_client: DspClient,
 ) -> list[Ap]:
     aps_to_delete = [ap for ap in existing_aps if ap.forGroup == forGroup]
     if not aps_to_delete:
         logger.warning(f"There are no APs to delete on {host} for group {forGroup}")
-        warnings.warn(f"There are no APs to delete on {host} for group {forGroup}")
         return existing_aps
-    print(f"Deleting the Administrative Permissions for group {forGroup} on server {host}")
     logger.info(f"Deleting the Administrative Permissions for group {forGroup} on server {host}")
     for ap in aps_to_delete:
-        _delete_ap_on_server(
-            ap=ap,
-            host=host,
-            token=token,
-        )
+        _delete_ap_on_server(ap, dsp_client)
         existing_aps.remove(ap)
-        logger.info(f"Deleted Administrative Permission {ap.iri} on host {host}")
+        logger.info(f"Deleted Administrative Permission {ap.iri}")
     return existing_aps

dsp_permissions_scripts/ap/ap_get.py

@@ -1,14 +1,11 @@
 from typing import Any
 from urllib.parse import quote_plus
 
-import requests
-
 from dsp_permissions_scripts.ap.ap_model import Ap, ApValue
-from dsp_permissions_scripts.models.api_error import ApiError
-from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.models.errors import ApiError
+from dsp_permissions_scripts.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
 from dsp_permissions_scripts.utils.project import get_project_iri_by_shortcode
-from dsp_permissions_scripts.utils.try_request import http_call_with_retry
 
 logger = get_logger(__name__)
 
@@ -38,41 +35,22 @@ def create_admin_route_object_from_ap(ap: Ap) -> dict[str, Any]:
     return ap_dict
 
 
-def _get_all_aps_of_project(
-    project_iri: str,
-    host: str,
-    token: str,
-) -> list[Ap]:
-    headers = {"Authorization": f"Bearer {token}"}
+def _get_all_aps_of_project(project_iri: str, dsp_client: DspClient) -> list[Ap]:
     project_iri = quote_plus(project_iri, safe="")
-    protocol = get_protocol(host)
-    url = f"{protocol}://{host}/admin/permissions/ap/{project_iri}"
-    response = http_call_with_retry(
-        action=lambda: requests.get(url, headers=headers, timeout=20),
-        err_msg=f"Could not get APs of project {project_iri}",
-    )
-    if response.status_code != 200:
-        raise ApiError(f"Could not get APs of project {project_iri}", response.text, response.status_code)
-    aps: list[dict[str, Any]] = response.json()["administrative_permissions"]
+    try:
+        response = dsp_client.get(f"/admin/permissions/ap/{project_iri}")
+    except ApiError as err:
+        err.message = f"Could not get APs of project {project_iri}"
+        raise err from None
+    aps: list[dict[str, Any]] = response["administrative_permissions"]
     ap_objects = [create_ap_from_admin_route_object(ap) for ap in aps]
     return ap_objects
 
 
-def get_aps_of_project(
-    host: str,
-    shortcode: str,
-    token: str,
-) -> list[Ap]:
+def get_aps_of_project(shortcode: str, dsp_client: DspClient) -> list[Ap]:
     """Returns the Administrative Permissions for a project."""
-    project_iri = get_project_iri_by_shortcode(
-        shortcode=shortcode,
-        host=host,
-    )
-    aps = _get_all_aps_of_project(
-        project_iri=project_iri,
-        host=host,
-        token=token,
-    )
-    print(f"Retrieved {len(aps)} Administrative Permissions of project {shortcode} on server {host}")
-    logger.info(f"Retrieved {len(aps)} Administrative Permissions of project {shortcode} on server {host}")
+    logger.info("****** Retrieving all Administrative Permissions... ******")
+    project_iri = get_project_iri_by_shortcode(shortcode, dsp_client)
+    aps = _get_all_aps_of_project(project_iri, dsp_client)
+    logger.info(f"Retrieved {len(aps)} Administrative Permissions")
     return aps

dsp_permissions_scripts/ap/ap_serialize.py

@@ -26,7 +26,6 @@ def serialize_aps_of_project(
     aps_as_dict = {explanation_string: aps_as_dicts}
     with open(filepath, mode="w", encoding="utf-8") as f:
         f.write(json.dumps(aps_as_dict, ensure_ascii=False, indent=2))
-    print(f"{len(project_aps)} APs have been written to file {str(filepath)}")
     logger.info(f"{len(project_aps)} APs have been written to file {str(filepath)}")
 
 

dsp_permissions_scripts/ap/ap_set.py

@@ -1,67 +1,40 @@
-import warnings
 from typing import Any
 from urllib.parse import quote_plus
 
-import requests
-
 from dsp_permissions_scripts.ap.ap_get import (
     create_admin_route_object_from_ap,
     create_ap_from_admin_route_object,
 )
 from dsp_permissions_scripts.ap.ap_model import Ap
-from dsp_permissions_scripts.models.api_error import ApiError
-from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.models.errors import ApiError
+from dsp_permissions_scripts.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
-from dsp_permissions_scripts.utils.try_request import http_call_with_retry
 
 logger = get_logger(__name__)
 
 
-def _update_ap_on_server(
-    ap: Ap,
-    host: str,
-    token: str,
-) -> Ap:
+def _update_ap_on_server(ap: Ap, dsp_client: DspClient) -> Ap:
     iri = quote_plus(ap.iri, safe="")
-    headers = {"Authorization": f"Bearer {token}"}
-    protocol = get_protocol(host)
-    url = f"{protocol}://{host}/admin/permissions/{iri}/hasPermissions"
     payload = {"hasPermissions": create_admin_route_object_from_ap(ap)["hasPermissions"]}
-    response = http_call_with_retry(
-        action=lambda: requests.put(url, headers=headers, json=payload, timeout=20),
-        err_msg=f"Could not update Administrative Permission {ap.iri}",
-    )
-    if response.status_code != 200:
-        raise ApiError(
-            message=f"Could not update Administrative Permission {ap.iri}",
-            response_text=response.text,
-            status_code=response.status_code,
-            payload=payload,
-        )
-    ap_updated: dict[str, Any] = response.json()["administrative_permission"]
+    try:
+        response = dsp_client.put(f"/admin/permissions/{iri}/hasPermissions", data=payload)
+    except ApiError as err:
+        err.message = f"Could not update Administrative Permission {ap.iri}"
+        raise err from None
+    ap_updated: dict[str, Any] = response["administrative_permission"]
     ap_object_updated = create_ap_from_admin_route_object(ap_updated)
     return ap_object_updated
 
 
-def apply_updated_aps_on_server(
-    aps: list[Ap],
-    host: str,
-    token: str,
-) -> None:
+def apply_updated_aps_on_server(aps: list[Ap], host: str, dsp_client: DspClient) -> None:
     if not aps:
         logger.warning(f"There are no APs to update on {host}")
-        warnings.warn(f"There are no APs to update on {host}")
         return
-    logger.info(f"Updating {len(aps)} APs on {host}...")
-    print(f"Updating {len(aps)} APs on {host}...")
+    logger.info(f"****** Updating {len(aps)} Administrative Permissions on {host}... ******")
     for ap in aps:
         try:
-            _ = _update_ap_on_server(
-                ap=ap,
-                host=host,
-                token=token,
-            )
+            _ = _update_ap_on_server(ap, dsp_client)
             logger.info(f"Successfully updated AP {ap.iri}")
         except ApiError as err:
             logger.error(err)
-            warnings.warn(err.message)
+    logger.info(f"Finished updating {len(aps)} Administrative Permissions on {host}")

dsp_permissions_scripts/doap/doap_get.py

@@ -1,17 +1,14 @@
 from typing import Any
 from urllib.parse import quote_plus
 
-import requests
-
 from dsp_permissions_scripts.doap.doap_model import Doap, DoapTarget, DoapTargetType
-from dsp_permissions_scripts.models.api_error import ApiError
-from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.models.errors import ApiError
+from dsp_permissions_scripts.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
 from dsp_permissions_scripts.utils.project import get_project_iri_by_shortcode
 from dsp_permissions_scripts.utils.scope_serialization import (
     create_scope_from_admin_route_object,
 )
-from dsp_permissions_scripts.utils.try_request import http_call_with_retry
 
 logger = get_logger(__name__)
 
@@ -36,22 +33,14 @@ def _filter_doaps_by_target(
     return filtered_doaps
 
 
-def _get_all_doaps_of_project(
-    project_iri: str,
-    host: str,
-    token: str,
-) -> list[Doap]:
-    headers = {"Authorization": f"Bearer {token}"}
+def _get_all_doaps_of_project(project_iri: str, dsp_client: DspClient) -> list[Doap]:
     project_iri = quote_plus(project_iri, safe="")
-    protocol = get_protocol(host)
-    url = f"{protocol}://{host}/admin/permissions/doap/{project_iri}"
-    response = http_call_with_retry(
-        action=lambda: requests.get(url, headers=headers, timeout=20),
-        err_msg=f"Error while getting DOAPs of project {project_iri}",
-    )
-    if response.status_code != 200:
-        raise ApiError(f"Error while getting DOAPs of project {project_iri}", response.text, response.status_code)
-    doaps: list[dict[str, Any]] = response.json()["default_object_access_permissions"]
+    try:
+        response = dsp_client.get(f"/admin/permissions/doap/{project_iri}")
+    except ApiError as err:
+        err.message = f"Error while getting DOAPs of project {project_iri}"
+        raise err from None
+    doaps: list[dict[str, Any]] = response["default_object_access_permissions"]
     doap_objects = [create_doap_from_admin_route_response(doap) for doap in doaps]
     return doap_objects
 
@@ -73,32 +62,24 @@ def create_doap_from_admin_route_response(permission: dict[str, Any]) -> Doap:
 
 
 def get_doaps_of_project(
-    host: str,
     shortcode: str,
-    token: str,
+    dsp_client: DspClient,
     target_type: DoapTargetType = DoapTargetType.ALL,
 ) -> list[Doap]:
     """
     Returns the DOAPs for a project.
     Optionally, select only the DOAPs that are related to either a group, or a resource class, or a property.
     By default, all DOAPs are returned, regardless of their target (target=all).
     """
-    project_iri = get_project_iri_by_shortcode(
-        shortcode=shortcode,
-        host=host,
-    )
-    doaps = _get_all_doaps_of_project(
-        project_iri=project_iri,
-        host=host,
-        token=token,
-    )
+    logger.info("****** Retrieving all DOAPs... ******")
+    project_iri = get_project_iri_by_shortcode(shortcode, dsp_client)
+    doaps = _get_all_doaps_of_project(project_iri, dsp_client)
     filtered_doaps = _filter_doaps_by_target(
         doaps=doaps,
         target=target_type,
     )
-    msg = f"Retrieved {len(doaps)} DOAPs of project {shortcode} on server {host}"
+    msg = f"Retrieved {len(doaps)} DOAPs"
     if target_type != DoapTargetType.ALL:
         msg += f", {len(filtered_doaps)} of which are related to {target_type}."
-    print(msg)
     logger.info(msg)
     return filtered_doaps

dsp_permissions_scripts/doap/doap_serialize.py

@@ -29,7 +29,6 @@ def serialize_doaps_of_project(
     doaps_as_dict = {explanation_string: doaps_as_dicts}
     with open(filepath, mode="w", encoding="utf-8") as f:
         f.write(json.dumps(doaps_as_dict, ensure_ascii=False, indent=2))
-    print(f"{len(project_doaps)} DOAPs have been written to file {str(filepath)}")
     logger.info(f"{len(project_doaps)} DOAPs have been written to file {str(filepath)}")
 
 

dsp_permissions_scripts/doap/doap_set.py

@@ -1,63 +1,39 @@
-import warnings
 from urllib.parse import quote_plus
 
-import requests
-
 from dsp_permissions_scripts.doap.doap_get import create_doap_from_admin_route_response
 from dsp_permissions_scripts.doap.doap_model import Doap
-from dsp_permissions_scripts.models.api_error import ApiError
+from dsp_permissions_scripts.models.errors import ApiError
 from dsp_permissions_scripts.models.scope import PermissionScope
-from dsp_permissions_scripts.utils.authentication import get_protocol
+from dsp_permissions_scripts.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
 from dsp_permissions_scripts.utils.scope_serialization import (
     create_admin_route_object_from_scope,
 )
-from dsp_permissions_scripts.utils.try_request import http_call_with_retry
 
 logger = get_logger(__name__)
 
 
-def _update_doap_scope_on_server(
-    doap_iri: str,
-    scope: PermissionScope,
-    host: str,
-    token: str,
-) -> Doap:
+def _update_doap_scope_on_server(doap_iri: str, scope: PermissionScope, dsp_client: DspClient) -> Doap:
     iri = quote_plus(doap_iri, safe="")
-    headers = {"Authorization": f"Bearer {token}"}
-    protocol = get_protocol(host)
-    url = f"{protocol}://{host}/admin/permissions/{iri}/hasPermissions"
     payload = {"hasPermissions": create_admin_route_object_from_scope(scope)}
-    response = http_call_with_retry(
-        action=lambda: requests.put(url, headers=headers, json=payload, timeout=20),
-        err_msg=f"Could not update scope of DOAP {doap_iri}",
-    )
-    if response.status_code != 200:
-        raise ApiError( f"Could not update scope of DOAP {doap_iri}", response.text, response.status_code, payload)
-    new_doap = create_doap_from_admin_route_response(response.json()["default_object_access_permission"])
+    try:
+        response = dsp_client.put(f"/admin/permissions/{iri}/hasPermissions", data=payload)
+    except ApiError as err:
+        err.message = f"Could not update scope of DOAP {doap_iri}"
+        raise err from None
+    new_doap = create_doap_from_admin_route_response(response["default_object_access_permission"])
     return new_doap
 
 
-def apply_updated_doaps_on_server(
-    doaps: list[Doap],
-    host: str,
-    token: str,
-) -> None:
+def apply_updated_doaps_on_server(doaps: list[Doap], host: str, dsp_client: DspClient) -> None:
     if not doaps:
         logger.warning(f"There are no DOAPs to update on {host}")
-        warnings.warn(f"There are no DOAPs to update on {host}")
         return
-    logger.info(f"Updating {len(doaps)} DOAPs on {host}...")
-    print(f"Updating {len(doaps)} DOAPs on {host}...")
+    logger.info(f"****** Updating {len(doaps)} DOAPs on {host}... ******")
     for d in doaps:
         try:
-            _ = _update_doap_scope_on_server(
-                doap_iri=d.doap_iri,
-                scope=d.scope,
-                host=host,
-                token=token,
-            )
+            _ = _update_doap_scope_on_server(d.doap_iri, d.scope, dsp_client)
             logger.info(f"Successfully updated DOAP {d.doap_iri}")
         except ApiError as err:
             logger.error(err)
-            warnings.warn(err.message)
+    logger.info(f"Finished updating {len(doaps)} DOAPs on {host}")