refactor: remove redundant parameter "host"

dsp_permissions_scripts/ap/ap_delete.py

@@ -19,16 +19,15 @@ def _delete_ap_on_server(ap: Ap, dsp_client: DspClient) -> None:
 
 
 def delete_ap_of_group_on_server(
-    host: str,
     existing_aps: list[Ap],
     forGroup: Group,
     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}")
+        logger.warning(f"There are no APs to delete on {dsp_client.server} for group {forGroup}")
         return existing_aps
-    logger.info(f"Deleting the Administrative Permissions for group {forGroup} on server {host}")
+    logger.info(f"Deleting the Administrative Permissions for group {forGroup} on server {dsp_client.server}")
     for ap in aps_to_delete:
         _delete_ap_on_server(ap, dsp_client)
         existing_aps.remove(ap)

dsp_permissions_scripts/ap/ap_serialize.py

@@ -17,12 +17,12 @@ def serialize_aps_of_project(
     project_aps: list[Ap],
     shortcode: str,
     mode: Literal["original", "modified"],
-    host: str,
+    server: str,
 ) -> None:
     """Serialize the APs of a project to a JSON file."""
     filepath = _get_file_path(shortcode, mode)
     filepath.parent.mkdir(parents=True, exist_ok=True)
-    explanation_string = f"{get_timestamp()}: Project {shortcode} on host {host} has {len(project_aps)} APs"
+    explanation_string = f"{get_timestamp()}: Project {shortcode} on server {server} has {len(project_aps)} APs"
     aps_as_dicts = [ap.model_dump(exclude_none=True, mode="json") for ap in project_aps]
     aps_as_dict = {explanation_string: aps_as_dicts}
     with open(filepath, mode="w", encoding="utf-8") as f:

dsp_permissions_scripts/ap/ap_set.py

@@ -27,18 +27,18 @@ def _update_ap_scope_on_server(ap: Ap, dsp_client: DspClient) -> Ap:
     return ap_object_updated
 
 
-def apply_updated_scopes_of_aps_on_server(aps: list[Ap], host: str, dsp_client: DspClient) -> None:
+def apply_updated_scopes_of_aps_on_server(aps: list[Ap], dsp_client: DspClient) -> None:
     if not aps:
-        logger.warning(f"There are no APs to update on {host}")
+        logger.warning(f"There are no APs to update on {dsp_client.server}")
         return
-    logger.info(f"****** Updating scopes of {len(aps)} Administrative Permissions on {host}... ******")
+    logger.info(f"****** Updating scopes of {len(aps)} Administrative Permissions on {dsp_client.server}... ******")
     for ap in aps:
         try:
             _ = _update_ap_scope_on_server(ap, dsp_client)
             logger.info(f"Successfully updated AP {ap.iri}")
         except ApiError as err:
             logger.error(err)
-    logger.info(f"Finished updating scopes of {len(aps)} Administrative Permissions on {host}")
+    logger.info(f"Finished updating scopes of {len(aps)} Administrative Permissions on {dsp_client.server}")
 
 
 def create_new_ap_on_server(

dsp_permissions_scripts/doap/doap_serialize.py

@@ -17,12 +17,12 @@ def serialize_doaps_of_project(
     project_doaps: list[Doap],
     shortcode: str,
     mode: Literal["original", "modified"],
-    host: str,
+    server: str,
 ) -> None:
     """Serialize the DOAPs of a project to a JSON file."""
     filepath = _get_file_path(shortcode, mode)
     filepath.parent.mkdir(parents=True, exist_ok=True)
-    explanation_string = f"{get_timestamp()}: Project {shortcode} on host {host} has {len(project_doaps)} DOAPs"
+    explanation_string = f"{get_timestamp()}: Project {shortcode} on server {server} has {len(project_doaps)} DOAPs"
     doaps_as_dicts = [doap.model_dump(exclude_none=True, mode="json") for doap in project_doaps]
     doaps_as_dict = {explanation_string: doaps_as_dicts}
     with open(filepath, mode="w", encoding="utf-8") as f:

dsp_permissions_scripts/doap/doap_set.py

@@ -26,18 +26,18 @@ def _update_doap_scope_on_server(doap_iri: str, scope: PermissionScope, dsp_clie
     return new_doap
 
 
-def apply_updated_scopes_of_doaps_on_server(doaps: list[Doap], host: str, dsp_client: DspClient) -> None:
+def apply_updated_scopes_of_doaps_on_server(doaps: list[Doap], dsp_client: DspClient) -> None:
     if not doaps:
-        logger.warning(f"There are no DOAPs to update on {host}")
+        logger.warning(f"There are no DOAPs to update on {dsp_client.server}")
         return
-    logger.info(f"****** Updating scopes of {len(doaps)} DOAPs on {host}... ******")
+    logger.info(f"****** Updating scopes of {len(doaps)} DOAPs on {dsp_client.server}... ******")
     for d in doaps:
         try:
             _ = _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)
-    logger.info(f"Finished updating scopes of {len(doaps)} DOAPs on {host}")
+    logger.info(f"Finished updating scopes of {len(doaps)} DOAPs on {dsp_client.server}")
 
 
 def create_new_doap_on_server(

dsp_permissions_scripts/oap/oap_set.py

@@ -117,11 +117,12 @@ def _update_batch(batch: tuple[ResourceOap | ValueOap, ...], dsp_client: DspClie
 def _write_failed_iris_to_file(
     failed_iris: list[str],
     shortcode: str,
-    host: str,
+    server: str,
     filename: str,
 ) -> None:
     with open(filename, "w", encoding="utf-8") as f:
-        f.write(f"Problems occurred while updating the OAPs of these resources (project {shortcode}, host {host}):\n")
+        msg = f"Problems occurred while updating the OAPs of these resources (project {shortcode}, server {server}):\n"
+        f.write(msg)
         f.write("\n".join(failed_iris))
 
 
@@ -137,7 +138,6 @@ def _launch_thread_pool(oaps: list[ResourceOap | ValueOap], nthreads: int, dsp_c
 
 def apply_updated_oaps_on_server(
     oaps: list[ResourceOap | ValueOap],
-    host: str,
     shortcode: str,
     dsp_client: DspClient,
     nthreads: int = 4,
@@ -147,11 +147,12 @@ def apply_updated_oaps_on_server(
     Don't forget to set a number of threads that doesn't overload the server.
     """
     if not oaps:
-        logger.warning(f"There are no OAPs to update on {host}")
+        logger.warning(f"There are no OAPs to update on {dsp_client.server}")
         return
     value_oap_count = sum(isinstance(oap, ValueOap) for oap in oaps)
     res_oap_count = sum(isinstance(oap, ResourceOap) for oap in oaps)
-    logger.info(f"******* Updating {res_oap_count} resource OAPs and {value_oap_count} value OAPs on {host}... *******")
+    msg = f"Updating {res_oap_count} resource OAPs and {value_oap_count} value OAPs on {dsp_client.server}..."
+    logger.info(f"******* {msg} *******")
 
     failed_iris = _launch_thread_pool(oaps, nthreads, dsp_client)
     if failed_iris:
@@ -160,9 +161,10 @@ def apply_updated_oaps_on_server(
         _write_failed_iris_to_file(
             failed_iris=sorted(failed_iris),
             shortcode=shortcode,
-            host=host,
+            server=dsp_client.server,
             filename=filename,
         )
         msg = f"ERROR: {len(failed_iris)} resources or values could not be updated. They were written to {filename}."
         logger.error(msg)
-    logger.info(f"Updated {res_oap_count} resource OAPs and {value_oap_count} value OAPs on {host}... *******")
+    msg = f"Updated {res_oap_count} resource OAPs and {value_oap_count} value OAPs on {dsp_client.server}... *******"
+    logger.info(msg)

dsp_permissions_scripts/template.py

@@ -68,17 +68,16 @@ def modify_oaps(oaps: list[Oap]) -> list[ResourceOap | ValueOap]:
     return modified_oaps
 
 
-def update_aps(host: str, shortcode: str, dsp_client: DspClient) -> None:
+def update_aps(shortcode: str, dsp_client: DspClient) -> None:
     """Sample function to modify the Administrative Permissions of a project."""
     project_aps = get_aps_of_project(shortcode, dsp_client)
     serialize_aps_of_project(
         project_aps=project_aps,
         shortcode=shortcode,
         mode="original",
-        host=host,
+        server=dsp_client.server,
     )
     remaining_aps = delete_ap_of_group_on_server(
-        host=host,
         existing_aps=project_aps,
         forGroup=group.PROJECT_MEMBER,
         dsp_client=dsp_client,
@@ -93,24 +92,24 @@ def update_aps(host: str, shortcode: str, dsp_client: DspClient) -> None:
     if not modified_aps:
         logger.info("There are no APs to update.")
         return
-    apply_updated_scopes_of_aps_on_server(modified_aps, host, dsp_client)
+    apply_updated_scopes_of_aps_on_server(modified_aps, dsp_client)
     project_aps_updated = get_aps_of_project(shortcode, dsp_client)
     serialize_aps_of_project(
         project_aps=project_aps_updated,
         shortcode=shortcode,
         mode="modified",
-        host=host,
+        server=dsp_client.server,
     )
 
 
-def update_doaps(host: str, shortcode: str, dsp_client: DspClient) -> None:
+def update_doaps(shortcode: str, dsp_client: DspClient) -> None:
     """Sample function to modify the Default Object Access Permissions of a project."""
     project_doaps = get_doaps_of_project(shortcode, dsp_client)
     serialize_doaps_of_project(
         project_doaps=project_doaps,
         shortcode=shortcode,
         mode="original",
-        host=host,
+        server=dsp_client.server,
     )
     _ = create_new_doap_on_server(
         target=NewGroupDoapTarget(group=group.CREATOR),
@@ -122,17 +121,17 @@ def update_doaps(host: str, shortcode: str, dsp_client: DspClient) -> None:
     if not project_doaps_modified:
         logger.info("There are no DOAPs to update.")
         return
-    apply_updated_scopes_of_doaps_on_server(project_doaps_modified, host, dsp_client)
+    apply_updated_scopes_of_doaps_on_server(project_doaps_modified, dsp_client)
     project_doaps_updated = get_doaps_of_project(shortcode, dsp_client)
     serialize_doaps_of_project(
         project_doaps=project_doaps_updated,
         shortcode=shortcode,
         mode="modified",
-        host=host,
+        server=dsp_client.server,
     )
 
 
-def update_oaps(host: str, shortcode: str, dsp_client: DspClient, oap_config: OapRetrieveConfig) -> None:
+def update_oaps(shortcode: str, dsp_client: DspClient, oap_config: OapRetrieveConfig) -> None:
     """Sample function to modify the Object Access Permissions of a project."""
     oaps = get_all_oaps_of_project(shortcode, dsp_client, oap_config)
     serialize_oaps(oaps, shortcode, mode="original")
@@ -142,7 +141,6 @@ def update_oaps(host: str, shortcode: str, dsp_client: DspClient, oap_config: Oa
         return
     apply_updated_oaps_on_server(
         oaps=oaps_modified,
-        host=host,
         shortcode=shortcode,
         dsp_client=dsp_client,
         nthreads=4,
@@ -172,17 +170,14 @@ def main() -> None:
     )
 
     update_aps(
-        host=host,
         shortcode=shortcode,
         dsp_client=dsp_client,
     )
     update_doaps(
-        host=host,
         shortcode=shortcode,
         dsp_client=dsp_client,
     )
     update_oaps(
-        host=host,
         shortcode=shortcode,
         dsp_client=dsp_client,
         oap_config=oap_config,

tests/test_ap_serialization.py

@@ -47,12 +47,12 @@ def test_serialize_aps_of_project(self) -> None:
             project_aps=[self.ap1, self.ap2],
             shortcode=self.shortcode,
             mode="original",
-            host=Hosts.LOCALHOST,
+            server=Hosts.LOCALHOST,
         )
         with open(self.output_file, mode="r", encoding="utf-8") as f:
             aps_file = json.load(f)
         explanation_text = next(iter(aps_file.keys()))
-        assert re.search(r"Project 1234 on host .+ has \d+ APs", explanation_text)
+        assert re.search(r"Project 1234 on server .+ has \d+ APs", explanation_text)
         aps_as_dicts = aps_file[explanation_text]
         assert self.ap1 == Ap.model_validate(aps_as_dicts[0])
         assert self.ap2 == Ap.model_validate(aps_as_dicts[1])

tests/test_doap_serialization.py

@@ -51,7 +51,7 @@ def test_doap_serialization(self) -> None:
             project_doaps=doaps_original,
             shortcode=self.shortcode,
             mode="original",
-            host=Hosts.LOCALHOST,
+            server=Hosts.LOCALHOST,
         )
         deserialized_doaps = deserialize_doaps_of_project(
             shortcode=self.shortcode,