chore: prepare next PR (renaming stuff) (#91)

dsp_permissions_scripts/oap/oap_get.py

@@ -16,25 +16,25 @@
 def _get_all_resource_oaps_of_resclass(resclass_iri: str, project_iri: str, dsp_client: DspClient) -> list[Oap]:
     logger.info(f"Getting all resource OAPs of class {resclass_iri}...")
     headers = {"X-Knora-Accept-Project": project_iri}
-    resources: list[Oap] = []
+    all_oaps: list[Oap] = []
     page = 0
     more = True
     while more:
         logger.info(f"Getting page {page}...")
         try:
-            more, iris = _get_next_page(
+            more, oaps = _get_next_page(
                 resclass_iri=resclass_iri,
                 page=page,
                 headers=headers,
                 dsp_client=dsp_client,
             )
-            resources.extend(iris)
+            all_oaps.extend(oaps)
             page += 1
         except ApiError as err:
             logger.error(f"{err}\nStop getting more pages, return what has been retrieved so far.")
             more = False
-    logger.info(f"Retrieved {len(resources)} resource OAPs of class {resclass_iri}")
-    return resources
+    logger.info(f"Retrieved {len(all_oaps)} OAPs of class {resclass_iri}")
+    return all_oaps
 
 
 def _get_next_page(
@@ -86,7 +86,7 @@ def get_resource(resource_iri: str, dsp_client: DspClient) -> dict[str, Any]:
         raise err from None
 
 
-def get_oap_by_resource_iri(resource_iri: str, dsp_client: DspClient) -> Oap:
+def get_resource_oap_by_iri(resource_iri: str, dsp_client: DspClient) -> Oap:
     resource = get_resource(resource_iri, dsp_client)
     scope = create_scope_from_string(resource["knora-api:hasPermissions"])
     return Oap(scope=scope, object_iri=resource_iri)
@@ -97,13 +97,13 @@ def get_all_resource_oaps_of_project(
     dsp_client: DspClient,
     excluded_class_iris: Iterable[str] = (),
 ) -> list[Oap]:
-    logger.info("******* Retrieving all resource OAPs... *******")
+    logger.info("******* Retrieving all OAPs... *******")
     project_iri, onto_iris = get_project_iri_and_onto_iris_by_shortcode(shortcode, dsp_client)
-    all_resource_oaps = []
     resclass_iris = get_all_resource_class_iris_of_project(onto_iris, dsp_client)
     resclass_iris = [x for x in resclass_iris if x not in excluded_class_iris]
+    all_oaps = []
     for resclass_iri in resclass_iris:
-        resource_oaps = _get_all_resource_oaps_of_resclass(resclass_iri, project_iri, dsp_client)
-        all_resource_oaps.extend(resource_oaps)
-    logger.info(f"Retrieved a TOTAL of {len(all_resource_oaps)} resource OAPs")
-    return all_resource_oaps
+        oaps = _get_all_resource_oaps_of_resclass(resclass_iri, project_iri, dsp_client)
+        all_oaps.extend(oaps)
+    logger.info(f"Retrieved a TOTAL of {len(all_oaps)} OAPs")
+    return all_oaps

dsp_permissions_scripts/oap/oap_serialize.py

@@ -12,33 +12,33 @@ def _get_project_data_path(shortcode: str, mode: Literal["original", "modified"]
     return Path(f"project_data/{shortcode}/OAPs_{mode}")
 
 
-def serialize_resource_oaps(
-    resource_oaps: list[Oap],
+def serialize_oaps(
+    oaps: list[Oap],
     shortcode: str,
     mode: Literal["original", "modified"],
 ) -> None:
-    """Serialize the resource OAPs to JSON files."""
-    if not resource_oaps:
+    """Serialize the OAPs to JSON files."""
+    if not oaps:
         logger.warning("No OAPs to serialize.")
         return
     folder = _get_project_data_path(shortcode, mode)
     folder.mkdir(parents=True, exist_ok=True)
-    logger.info(f"Writing {len(resource_oaps)} OAPs into {folder}")
-    for res_oap in resource_oaps:
+    logger.info(f"Writing {len(oaps)} OAPs into {folder}")
+    for res_oap in oaps:
         filename = re.sub(r"http://rdfh\.ch/[^/]+/", "resource_", res_oap.object_iri)
         with open(folder / f"{filename}.json", mode="w", encoding="utf-8") as f:
             f.write(res_oap.model_dump_json(indent=2))
-    logger.info(f"Successfully wrote {len(resource_oaps)} OAPs into {folder}")
+    logger.info(f"Successfully wrote {len(oaps)} OAPs into {folder}")
 
 
-def deserialize_resource_oaps(
+def deserialize_oaps(
     shortcode: str,
     mode: Literal["original", "modified"],
 ) -> list[Oap]:
     """Deserialize the resource OAPs from JSON files."""
     folder = _get_project_data_path(shortcode, mode)
-    resource_oaps = []
+    oaps = []
     for file in [f for f in folder.iterdir() if f.suffix == ".json"]:
         with open(file, mode="r", encoding="utf-8") as f:
-            resource_oaps.append(Oap.model_validate_json(f.read()))
-    return resource_oaps
+            oaps.append(Oap.model_validate_json(f.read()))
+    return oaps

dsp_permissions_scripts/oap/oap_set.py

@@ -57,12 +57,12 @@ def _update_permissions_for_value(  # noqa: PLR0913
     }
     try:
         dsp_client.put("/v2/values", data=payload)
+        logger.info(f"Updated permissions of resource {resource_iri}, value {value.value_iri}")
     except PermissionsAlreadyUpToDate:
         logger.warning(f"Permissions of resource {resource_iri}, value {value.value_iri} are already up to date")
     except ApiError as err:
         err.message = f"Error while updating permissions of resource {resource_iri}, value {value.value_iri}"
         raise err from None
-    logger.info(f"Updated permissions of resource {resource_iri}, value {value.value_iri}")
 
 
 def _update_permissions_for_resource(  # noqa: PLR0913
@@ -84,10 +84,12 @@ def _update_permissions_for_resource(  # noqa: PLR0913
         payload["knora-api:lastModificationDate"] = lmd
     try:
         dsp_client.put("/v2/resources", data=payload)
+        logger.info(f"Updated permissions of resource {resource_iri}")
+    except PermissionsAlreadyUpToDate:
+        logger.warning(f"Permissions of resource {resource_iri} are already up to date")
     except ApiError as err:
         err.message = f"ERROR while updating permissions of resource {resource_iri}"
         raise err from None
-    logger.info(f"Updated permissions of resource {resource_iri}")
 
 
 def _update_permissions_for_resource_and_values(
@@ -171,7 +173,7 @@ def _launch_thread_pool(resource_oaps: list[Oap], nthreads: int, dsp_client: Dsp
 
 
 def apply_updated_oaps_on_server(
-    resource_oaps: list[Oap],
+    oaps: list[Oap],
     host: str,
     shortcode: str,
     dsp_client: DspClient,
@@ -181,12 +183,12 @@ def apply_updated_oaps_on_server(
     Applies modified Object Access Permissions of resources (and their values) on a DSP server.
     Don't forget to set a number of threads that doesn't overload the server.
     """
-    if not resource_oaps:
+    if not oaps:
         logger.warning(f"There are no OAPs to update on {host}")
         return
-    logger.info(f"******* Updating OAPs of {len(resource_oaps)} resources on {host}... *******")
+    logger.info(f"******* Updating OAPs of {len(oaps)} resources on {host}... *******")
 
-    failed_res_iris = _launch_thread_pool(resource_oaps, nthreads, dsp_client)
+    failed_res_iris = _launch_thread_pool(oaps, nthreads, dsp_client)
 
     if failed_res_iris:
         timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
@@ -202,4 +204,4 @@ def apply_updated_oaps_on_server(
             f"They were written to {filename}."
         )
         logger.error(msg)
-    logger.info(f"Updated OAPs of {len(resource_oaps)} resources on {host}")
+    logger.info(f"Updated OAPs of {len(oaps)} resources on {host}")

dsp_permissions_scripts/template.py

@@ -13,7 +13,7 @@
 from dsp_permissions_scripts.models.scope import PUBLIC
 from dsp_permissions_scripts.oap.oap_get import get_all_resource_oaps_of_project
 from dsp_permissions_scripts.oap.oap_model import Oap
-from dsp_permissions_scripts.oap.oap_serialize import serialize_resource_oaps
+from dsp_permissions_scripts.oap.oap_serialize import serialize_oaps
 from dsp_permissions_scripts.oap.oap_set import apply_updated_oaps_on_server
 from dsp_permissions_scripts.utils.authentication import login
 from dsp_permissions_scripts.utils.dsp_client import DspClient
@@ -102,17 +102,17 @@ def update_doaps(host: str, shortcode: str, dsp_client: DspClient) -> None:
 def update_oaps(host: str, shortcode: str, dsp_client: DspClient) -> None:
     """Sample function to modify the Object Access Permissions of a project."""
     resource_oaps = get_all_resource_oaps_of_project(shortcode, dsp_client)
-    serialize_resource_oaps(resource_oaps, shortcode, mode="original")
+    serialize_oaps(resource_oaps, shortcode, mode="original")
     resource_oaps_modified = modify_oaps(oaps=resource_oaps)
     apply_updated_oaps_on_server(
-        resource_oaps=resource_oaps_modified,
+        oaps=resource_oaps_modified,
         host=host,
         shortcode=shortcode,
         dsp_client=dsp_client,
         nthreads=4,
     )
     resource_oaps_updated = get_all_resource_oaps_of_project(shortcode, dsp_client)
-    serialize_resource_oaps(resource_oaps_updated, shortcode, mode="modified")
+    serialize_oaps(resource_oaps_updated, shortcode, mode="modified")
 
 
 def main() -> None:

pyproject.toml

@@ -66,7 +66,6 @@ show_column_numbers = true
 strict = true
 disallow_untyped_defs = true
 plugins = ["pydantic.mypy"]
-follow_imports = "silent"
 warn_redundant_casts = true
 warn_unused_ignores = true
 disallow_any_generics = true

tests/test_oap_serialization.py

@@ -5,12 +5,13 @@
 from dsp_permissions_scripts.models import group
 from dsp_permissions_scripts.models.scope import PermissionScope
 from dsp_permissions_scripts.oap.oap_model import Oap
-from dsp_permissions_scripts.oap.oap_serialize import deserialize_resource_oaps
-from dsp_permissions_scripts.oap.oap_serialize import serialize_resource_oaps
+from dsp_permissions_scripts.oap.oap_serialize import deserialize_oaps
+from dsp_permissions_scripts.oap.oap_serialize import serialize_oaps
 from tests.test_scope_serialization import compare_scopes
 
 # ruff: noqa: PT009 (pytest-unittest-assertion) (remove this line when pytest is used instead of unittest)
 
+
 class TestOapSerialization(unittest.TestCase):
     shortcode = "1234"
 
@@ -34,12 +35,12 @@ def test_oap_serialization(self) -> None:
             ),
             object_iri=f"http://rdfh.ch/{self.shortcode}/resource-2",
         )
-        serialize_resource_oaps(
-            resource_oaps=[oap1, oap2],
+        serialize_oaps(
+            oaps=[oap1, oap2],
             shortcode=self.shortcode,
             mode="original",
         )
-        deserialized_oaps = deserialize_resource_oaps(
+        deserialized_oaps = deserialize_oaps(
             shortcode=self.shortcode,
             mode="original",
         )