chore: improve error handling

dsp_permissions_scripts/ap/ap_get.py

@@ -4,6 +4,7 @@
 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.utils.get_logger import get_logger
 from dsp_permissions_scripts.utils.project import get_project_iri_by_shortcode
@@ -47,7 +48,8 @@ def _get_all_aps_of_project(
     protocol = get_protocol(host)
     url = f"{protocol}://{host}/admin/permissions/ap/{project_iri}"
     response = requests.get(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError("Could not get APs of project", response.text, response.status_code)
     aps: list[dict[str, Any]] = response.json()["administrative_permissions"]
     ap_objects = [create_ap_from_admin_route_object(ap) for ap in aps]
     return ap_objects

dsp_permissions_scripts/ap/ap_set.py

@@ -9,6 +9,7 @@
     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.utils.get_logger import get_logger, get_timestamp
 
@@ -34,7 +35,8 @@ def _delete_single_ap(
     protocol = get_protocol(host)
     url = f"{protocol}://{host}/admin/permissions/{ap_iri}"
     response = requests.delete(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError(f"Could not delete Administrative Permission {ap.iri}", response.text, response.status_code)
     logger.info(f"Deleted Administrative Permission {ap.iri} on host {host}")
 
 
@@ -52,7 +54,13 @@ def _update_ap(
     url = f"{protocol}://{host}/admin/permissions/{iri}/hasPermissions"
     payload = {"hasPermissions": create_admin_route_object_from_ap(ap)["hasPermissions"]}
     response = requests.put(url, headers=headers, json=payload, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    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"]
     ap_object_updated = create_ap_from_admin_route_object(ap_updated)
     return ap_object_updated
@@ -90,9 +98,9 @@ def apply_updated_aps_on_server(
                 token=token,
             )
             _log_and_print_ap_update(ap=new_ap)
-        except Exception:  # pylint: disable=broad-exception-caught
-            logger.error(f"ERROR while updating Administrative Permission {ap.iri}", exc_info=True)
-            warnings.warn(f"ERROR while updating Administrative Permission {ap.iri}")
+        except ApiError as err:
+            logger.error(err)
+            warnings.warn(err.message)
 
     print(f"{get_timestamp()}: All APs have been updated.")
 

dsp_permissions_scripts/doap/doap_get.py

@@ -4,6 +4,7 @@
 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.utils.get_logger import get_logger
 from dsp_permissions_scripts.utils.project import get_project_iri_by_shortcode
@@ -47,7 +48,8 @@ def _get_all_doaps_of_project(
     protocol = get_protocol(host)
     url = f"{protocol}://{host}/admin/permissions/doap/{project_iri}"
     response = requests.get(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    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"]
     doap_objects = [create_doap_from_admin_route_response(doap) for doap in doaps]
     return doap_objects

dsp_permissions_scripts/doap/doap_set.py

@@ -5,6 +5,7 @@
 
 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.scope import PermissionScope
 from dsp_permissions_scripts.utils.authentication import get_protocol
 from dsp_permissions_scripts.utils.get_logger import get_logger, get_timestamp
@@ -30,7 +31,8 @@ def _update_doap_scope(
     url = f"{protocol}://{host}/admin/permissions/{iri}/hasPermissions"
     payload = {"hasPermissions": create_admin_route_object_from_scope(scope)}
     response = requests.put(url, headers=headers, json=payload, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    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"])
     return new_doap
 
@@ -68,8 +70,8 @@ def apply_updated_doaps_on_server(
                 token=token,
             )
             _log_and_print_doap_update(doap=new_doap)
-        except Exception:  # pylint: disable=broad-exception-caught
-            logger.error(f"ERROR while updating DOAP {d.doap_iri}", exc_info=True)
-            warnings.warn(f"ERROR while updating DOAP {d.doap_iri}")
+        except ApiError as err:
+            logger.error(err)
+            warnings.warn(err.message)
 
     print(f"{get_timestamp()}: All DOAPs have been updated.")

dsp_permissions_scripts/models/api_error.py

@@ -0,0 +1,16 @@
+import pprint
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+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))

dsp_permissions_scripts/oap/oap_get_set.py

@@ -1,12 +1,12 @@
 # pylint: disable=too-many-arguments
 
-import json
 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_model import Oap
@@ -50,7 +50,8 @@ def _get_resource(
     url = f"{protocol}://{host}/v2/resources/{iri}"
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.get(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    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
 
@@ -109,11 +110,11 @@ def _update_permissions_for_value(
             msg = f"Permissions of resource {resource_iri}, value {value.value_iri} are already up to date"
             logger.warning(msg)
     elif response.status_code != 200:
-        logger.error(
-            f"Error while updating permissions of resource {resource_iri}, value {value.value_iri}. "
-            f"Response status code: {response.status_code}. "
-            f"Response text: {response.text}. "
-            f"Payload: {json.dumps(payload, indent=4)}"
+        raise ApiError(
+            message=f"Error while updating permissions of resource {resource_iri}, value {value.value_iri}",
+            response_text=response.text, 
+            status_code=response.status_code, 
+            payload=payload
         )
     else:
         logger.info(f"Updated permissions of resource {resource_iri}, value {value.value_iri}")
@@ -143,7 +144,13 @@ def _update_permissions_for_resource(
     url = f"{protocol}://{host}/v2/resources"
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.put(url, headers=headers, json=payload, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError(
+            message=f"ERROR while updating permissions of resource {resource_iri}",
+            response_text=response.text,
+            status_code=response.status_code, 
+            payload=payload, 
+        )
     logger.info(f"Updated permissions of resource {resource_iri}")
 
 
@@ -214,9 +221,9 @@ def apply_updated_oaps_on_server(
                 host=host,
                 token=token,
             )
-        except Exception:  # pylint: disable=broad-exception-caught
-            logger.error(f"ERROR while updating permissions of resource {resource_oap.object_iri}", exc_info=True)
-            warnings.warn(f"ERROR while updating permissions of resource {resource_oap.object_iri}")
+        except ApiError as err:
+            logger.error(err)
+            warnings.warn(err.message)
             failed_res_iris.append(resource_oap.object_iri)
         logger.info(f"Updated permissions of resource {resource_oap.object_iri} and its values.")
 

dsp_permissions_scripts/utils/authentication.py

@@ -2,6 +2,8 @@
 
 import requests
 
+from dsp_permissions_scripts.models.api_error import ApiError
+
 
 def _get_token(host: str, email: str, pw: str) -> str:
     """
@@ -10,7 +12,8 @@ def _get_token(host: str, email: str, pw: str) -> str:
     protocol = get_protocol(host)
     url = f"{protocol}://{host}/v2/authentication"
     response = requests.post(url, json={"email": email, "password": pw}, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError("Could not login", response.text, response.status_code)
     token: str = response.json()["token"]
     return token
 

dsp_permissions_scripts/utils/project.py

@@ -1,7 +1,9 @@
+import warnings
 from urllib.parse import quote_plus
 
 import requests
 
+from dsp_permissions_scripts.models.api_error import ApiError
 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, get_timestamp
@@ -43,7 +45,8 @@ def _get_onto_iris_of_project(
     url = f"{protocol}://{host}/v2/ontologies/metadata"
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.get(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError("Could not get onto IRIs", response.text, response.status_code)
     all_ontologies = response.json().get("@graph")
     project_onto_iris = [o["@id"] for o in all_ontologies if o["knora-api:attachedToProject"]["@id"] == project_iri]
     return project_onto_iris
@@ -58,7 +61,8 @@ def _get_class_iris_of_onto(
     url = f"{protocol}://{host}/v2/ontologies/allentities/{quote_plus(onto_iri)}"
     headers = {"Authorization": f"Bearer {token}"}
     response = requests.get(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError("Could not get class IRIs", response.text, response.status_code)
     all_entities = response.json()["@graph"]
     context = response.json()["@context"]
     class_ids = [c["@id"] for c in all_entities if c.get("knora-api:isResourceClass")]
@@ -81,15 +85,20 @@ def _get_all_resource_oaps_of_resclass(
     more = True
     while more:
         logger.info(f"Getting page {page}...")
-        more, iris = _get_next_page(
-            protocol=protocol,
-            host=host,
-            resclass_iri=resclass_iri,
-            page=page,
-            headers=headers,
-        )
-        resources.extend(iris)
-        page += 1
+        try:
+            more, iris = _get_next_page(
+                protocol=protocol,
+                host=host,
+                resclass_iri=resclass_iri,
+                page=page,
+                headers=headers,
+            )
+            resources.extend(iris)
+            page += 1
+        except ApiError as err:
+            logger.error(f"{err}\nStop getting more pages, return what has been retrieved so far.")
+            warnings.warn(f"{err.message}\nStop getting more pages, return what has been retrieved so far.")
+            more = False
     print(f"{get_timestamp()}: Retrieved {len(resources)} resource OAPs of class {resclass_iri}.")
     logger.info(f"Retrieved {len(resources)} resource OAPs of class {resclass_iri}.")
     return resources
@@ -113,7 +122,8 @@ def _get_next_page(
     """
     url = f"{protocol}://{host}/v2/resources?resourceClass={quote_plus(resclass_iri)}&page={page}"
     response = requests.get(url, headers=headers, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError("Could not get next page", response.text, response.status_code)
     result = response.json()
     if "@graph" in result:
         # result contains several resources: return them, then continue with next page
@@ -138,7 +148,8 @@ def get_project_iri_by_shortcode(shortcode: str, host: str) -> str:
     protocol = get_protocol(host)
     url = f"{protocol}://{host}/admin/projects/shortcode/{shortcode}"
     response = requests.get(url, timeout=5)
-    assert response.status_code == 200, f"Status {response.status_code}. Error message from DSP-API: {response.text}"
+    if response.status_code != 200:
+        raise ApiError("Cannot retrieve project IRI", response.text, response.status_code)
     iri: str = response.json()["project"]["id"]
     return iri