feat: log every request to DSP-API (#76)

dsp_permissions_scripts/ap/ap_delete.py

@@ -1,50 +1,35 @@
 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}")
         return existing_aps
     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}")
     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."""
     logger.info("****** Retrieving all Administrative Permissions... ******")
-    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,
-    )
+    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_set.py

@@ -1,63 +1,39 @@
 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}")
         return
     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)

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,9 +62,8 @@ 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]:
     """
@@ -84,15 +72,8 @@ def get_doaps_of_project(
     By default, all DOAPs are returned, regardless of their target (target=all).
     """
     logger.info("****** Retrieving all DOAPs... ******")
-    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,
-    )
+    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,

dsp_permissions_scripts/doap/doap_set.py

@@ -1,59 +1,38 @@
 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}")
         return
     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)

dsp_permissions_scripts/models/api_error.py → dsp_permissions_scripts/models/errors.py

@@ -1,16 +1,19 @@
 import pprint
-from dataclasses import dataclass, field
-from typing import Any
+from dataclasses import dataclass
 
 
-@dataclass(frozen=True)
+@dataclass
 class ApiError(Exception):
     """Exception raised when an error occurs while calling DSP-API."""
 
     message: str
     response_text: str | None = None
     status_code: int | None = None
-    payload: dict[str, Any] = field(default_factory=dict)
 
     def __str__(self) -> str:
         return pprint.pformat(vars(self))
+
+
+@dataclass
+class PermissionsAlreadyUpToDate(Exception):
+    message: str = "The submitted permissions are the same as the current ones"