Merge branch 'main' into wip/RDU-12-delete-doaps

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/models/errors.py

@@ -39,11 +39,6 @@ class SpecifiedPropsNotEmptyError(Exception):
     message: str = "specified_props must be empty if retrieve_values is not 'specified_props'"
 
 
-@dataclass
-class OapEmptyError(Exception):
-    message: str = "An OAP must specify at least one resource_oap or one value_oap"
-
-
 @dataclass
 class EmptyScopeError(Exception):
     message: str = "PermissionScope must not be empty"

dsp_permissions_scripts/models/host.py

@@ -3,10 +3,7 @@ class Hosts:
 
     LOCALHOST = "http://0.0.0.0:3333"
     PROD = "https://api.dasch.swiss"
-    RDU_STAGE = "https://api.rdu-stage.dasch.swiss"
     DEV = "https://api.dev.dasch.swiss"
-    LS_PROD = "https://api.ls-prod.admin.ch"
-    STAGE = "https://api.stage.dasch.swiss"
 
     @staticmethod
     def get_host(identifier: str) -> str:

dsp_permissions_scripts/oap/oap_get.py

@@ -31,6 +31,7 @@
     "knora-api:attachedToUser",
     "knora-api:userHasPermission",
     "knora-api:hasPermissions",
+    "knora-api:hasStandoffLinkToValue",
 ]
 KB_RESCLASSES = [
     "knora-api:VideoSegment",

dsp_permissions_scripts/oap/oap_model.py

@@ -4,9 +4,9 @@
 
 from pydantic import BaseModel
 from pydantic import ConfigDict
+from pydantic import Field
 from pydantic import model_validator
 
-from dsp_permissions_scripts.models.errors import OapEmptyError
 from dsp_permissions_scripts.models.errors import SpecifiedPropsEmptyError
 from dsp_permissions_scripts.models.errors import SpecifiedPropsNotEmptyError
 from dsp_permissions_scripts.models.errors import SpecifiedResClassesEmptyError
@@ -23,11 +23,18 @@ class Oap(BaseModel):
     resource_oap: ResourceOap
     value_oaps: list[ValueOap]
 
-    @model_validator(mode="after")
-    def check_consistency(self) -> Oap:
-        if not self.resource_oap and not self.value_oaps:
-            raise OapEmptyError()
-        return self
+
+class ModifiedOap(BaseModel):
+    """
+    Model representing a modified object access permission of a resource and its values.
+    This model is used to represent only the modified parts of an OAP, so it can be incomplete.
+    """
+
+    resource_oap: ResourceOap | None = None
+    value_oaps: list[ValueOap] = Field(default_factory=list)
+
+    def is_empty(self) -> bool:
+        return not (self.resource_oap or self.value_oaps)
 
 
 class ResourceOap(BaseModel):

dsp_permissions_scripts/oap/oap_set.py

@@ -8,7 +8,7 @@
 from dsp_permissions_scripts.models.errors import PermissionsAlreadyUpToDate
 from dsp_permissions_scripts.models.group import KNORA_ADMIN_ONTO_NAMESPACE
 from dsp_permissions_scripts.models.scope import PermissionScope
-from dsp_permissions_scripts.oap.oap_model import ResourceOap
+from dsp_permissions_scripts.oap.oap_model import ModifiedOap
 from dsp_permissions_scripts.oap.oap_model import ValueOap
 from dsp_permissions_scripts.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
@@ -72,60 +72,60 @@ def _update_permissions_for_resource(  # noqa: PLR0913
         raise err from None
 
 
-def _update_batch(batch: tuple[ResourceOap | ValueOap, ...], dsp_client: DspClient) -> list[str]:
+def _update_batch(batch: tuple[ModifiedOap, ...], dsp_client: DspClient) -> list[str]:
     failed_iris = []
     for oap in batch:
+        res_iri = oap.resource_oap.resource_iri if oap.resource_oap else oap.value_oaps[0].resource_iri
         try:
-            resource = dsp_client.get(f"/v2/resources/{quote_plus(oap.resource_iri, safe='')}")
+            resource = dsp_client.get(f"/v2/resources/{quote_plus(res_iri, safe='')}")
         except ApiError as exc:
             logger.error(
-                f"Cannot update resource {oap.resource_iri}. "
+                f"Cannot update resource {res_iri}. "
                 f"The resource cannot be retrieved for the following reason: {exc.message}"
             )
-            failed_iris.append(oap.resource_iri)
+            failed_iris.append(res_iri)
             continue
-        if isinstance(oap, ResourceOap):
+        if oap.resource_oap:
             try:
                 _update_permissions_for_resource(
-                    resource_iri=oap.resource_iri,
+                    resource_iri=oap.resource_oap.resource_iri,
                     lmd=resource.get("knora-api:lastModificationDate"),
                     resource_type=resource["@type"],
                     context=resource["@context"] | {"knora-admin": KNORA_ADMIN_ONTO_NAMESPACE},
-                    scope=oap.scope,
+                    scope=oap.resource_oap.scope,
                     dsp_client=dsp_client,
                 )
             except ApiError as err:
                 logger.error(err)
-                failed_iris.append(oap.resource_iri)
-        elif isinstance(oap, ValueOap):
+                failed_iris.append(oap.resource_oap.resource_iri)
+        for val_oap in oap.value_oaps:
             try:
                 _update_permissions_for_value(
-                    resource_iri=oap.resource_iri,
-                    value=oap,
+                    resource_iri=val_oap.resource_iri,
+                    value=val_oap,
                     resource_type=resource["@type"],
                     context=resource["@context"] | {"knora-admin": KNORA_ADMIN_ONTO_NAMESPACE},
                     dsp_client=dsp_client,
                 )
             except ApiError as err:
                 logger.error(err)
-                failed_iris.append(oap.value_iri)
-        else:
-            raise ValueError(f"The provided OAP is neither a resource OAP nor a value OAP: {oap}")
+                failed_iris.append(val_oap.value_iri)
     return failed_iris
 
 
 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))
 
 
-def _launch_thread_pool(oaps: list[ResourceOap | ValueOap], nthreads: int, dsp_client: DspClient) -> list[str]:
+def _launch_thread_pool(oaps: list[ModifiedOap], nthreads: int, dsp_client: DspClient) -> list[str]:
     all_failed_iris: list[str] = []
     with ThreadPoolExecutor(max_workers=nthreads) as pool:
         jobs = [pool.submit(_update_batch, batch, dsp_client) for batch in itertools.batched(oaps, 100)]
@@ -136,8 +136,7 @@ def _launch_thread_pool(oaps: list[ResourceOap | ValueOap], nthreads: int, dsp_c
 
 
 def apply_updated_oaps_on_server(
-    oaps: list[ResourceOap | ValueOap],
-    host: str,
+    oaps: list[ModifiedOap],
     shortcode: str,
     dsp_client: DspClient,
     nthreads: int = 4,
@@ -146,12 +145,14 @@ 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.
     """
+    oaps = [oap for oap in oaps if oap.resource_oap or oap.value_oaps]
     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}... *******")
+    value_oap_count = sum(len(oap.value_oaps) for oap in oaps)
+    res_oap_count = sum(1 if oap.resource_oap else 0 for oap in oaps)
+    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)