Merge branch 'main' into wip/DEV-4061-update-list-of-iris

dsp_permissions_scripts/doap/doap_delete.py

@@ -0,0 +1,38 @@
+from urllib.parse import quote_plus
+
+from dsp_permissions_scripts.doap.doap_model import Doap
+from dsp_permissions_scripts.doap.doap_model import GroupDoapTarget
+from dsp_permissions_scripts.models.errors import ApiError
+from dsp_permissions_scripts.models.group import Group
+from dsp_permissions_scripts.utils.dsp_client import DspClient
+from dsp_permissions_scripts.utils.get_logger import get_logger
+
+logger = get_logger(__name__)
+
+
+def _delete_doap_on_server(doap: Doap, dsp_client: DspClient) -> None:
+    doap_iri = quote_plus(doap.doap_iri, safe="")
+    try:
+        dsp_client.delete(f"/admin/permissions/{doap_iri}")
+    except ApiError as err:
+        err.message = f"Could not delete DOAP {doap.doap_iri}"
+        raise err from None
+
+
+def delete_doap_of_group_on_server(
+    existing_doaps: list[Doap],
+    forGroup: Group,
+    dsp_client: DspClient,
+) -> list[Doap]:
+    doaps_to_delete = [
+        doap for doap in existing_doaps if isinstance(doap.target, GroupDoapTarget) and doap.target.group == forGroup
+    ]
+    if not doaps_to_delete:
+        logger.warning(f"There are no DOAPs to delete on {dsp_client.server} for group {forGroup}")
+        return existing_doaps
+    logger.info(f"Deleting the DOAP for group {forGroup} on server {dsp_client.server}")
+    for doap in doaps_to_delete:
+        _delete_doap_on_server(doap, dsp_client)
+        existing_doaps.remove(doap)
+        logger.info(f"Deleted DOAP {doap.doap_iri}")
+    return existing_doaps

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/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,45 +72,44 @@ 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
 
 
@@ -126,7 +125,7 @@ def _write_failed_iris_to_file(
         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)]
@@ -137,7 +136,7 @@ def _launch_thread_pool(oaps: list[ResourceOap | ValueOap], nthreads: int, dsp_c
 
 
 def apply_updated_oaps_on_server(
-    oaps: list[ResourceOap | ValueOap],
+    oaps: list[ModifiedOap],
     shortcode: str,
     dsp_client: DspClient,
     nthreads: int = 4,
@@ -146,11 +145,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.
     """
+    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 {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)
+    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} *******")
 

dsp_permissions_scripts/template.py

@@ -7,6 +7,7 @@
 from dsp_permissions_scripts.ap.ap_serialize import serialize_aps_of_project
 from dsp_permissions_scripts.ap.ap_set import apply_updated_scopes_of_aps_on_server
 from dsp_permissions_scripts.ap.ap_set import create_new_ap_on_server
+from dsp_permissions_scripts.doap.doap_delete import delete_doap_of_group_on_server
 from dsp_permissions_scripts.doap.doap_get import get_doaps_of_project
 from dsp_permissions_scripts.doap.doap_model import Doap
 from dsp_permissions_scripts.doap.doap_model import GroupDoapTarget
@@ -19,10 +20,9 @@
 from dsp_permissions_scripts.models.scope import OPEN
 from dsp_permissions_scripts.models.scope import PermissionScope
 from dsp_permissions_scripts.oap.oap_get import get_all_oaps_of_project
+from dsp_permissions_scripts.oap.oap_model import ModifiedOap
 from dsp_permissions_scripts.oap.oap_model import Oap
 from dsp_permissions_scripts.oap.oap_model import OapRetrieveConfig
-from dsp_permissions_scripts.oap.oap_model import ResourceOap
-from dsp_permissions_scripts.oap.oap_model import ValueOap
 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
@@ -54,17 +54,18 @@ def modify_doaps(doaps: list[Doap]) -> list[Doap]:
     return modified_doaps
 
 
-def modify_oaps(oaps: list[Oap]) -> list[ResourceOap | ValueOap]:
+def modify_oaps(oaps: list[Oap]) -> list[ModifiedOap]:
     """Adapt this sample to your needs."""
-    modified_oaps: list[ResourceOap | ValueOap] = []
+    modified_oaps: list[ModifiedOap] = []
     for oap in copy.deepcopy(oaps):
-        if group.SYSTEM_ADMIN not in oap.resource_oap.scope.CR:
-            oap.resource_oap.scope = oap.resource_oap.scope.add("CR", group.SYSTEM_ADMIN)
-            modified_oaps.append(oap.resource_oap)
+        new_oap = ModifiedOap()
+        if oap.resource_oap.scope != OPEN:
+            new_oap.resource_oap = oap.resource_oap.model_copy(update={"scope": OPEN})
         for value_oap in oap.value_oaps:
-            if group.SYSTEM_ADMIN not in value_oap.scope.CR:
-                value_oap.scope = value_oap.scope.add("CR", group.SYSTEM_ADMIN)
-                modified_oaps.append(value_oap)
+            if value_oap.scope != OPEN:
+                new_oap.value_oaps.append(value_oap.model_copy(update={"scope": OPEN}))
+        if not new_oap.is_empty():
+            modified_oaps.append(new_oap)
     return modified_oaps
 
 
@@ -111,13 +112,18 @@ def update_doaps(shortcode: str, dsp_client: DspClient) -> None:
         mode="original",
         server=dsp_client.server,
     )
+    remaining_doaps = delete_doap_of_group_on_server(
+        existing_doaps=project_doaps,
+        forGroup=group.PROJECT_MEMBER,
+        dsp_client=dsp_client,
+    )
     _ = create_new_doap_on_server(
         target=NewGroupDoapTarget(group=group.CREATOR),
         shortcode=shortcode,
         scope=PermissionScope.create(CR=[group.SYSTEM_ADMIN]),
         dsp_client=dsp_client,
     )
-    project_doaps_modified = modify_doaps(doaps=project_doaps)
+    project_doaps_modified = modify_doaps(doaps=remaining_doaps)
     if not project_doaps_modified:
         logger.info("There are no DOAPs to update.")
         return