edit

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_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
 
 
@@ -125,7 +124,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)]
@@ -136,7 +135,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],
     host: str,
     shortcode: str,
     dsp_client: DspClient,
@@ -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 {host}")
         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)
     logger.info(f"******* Updating {res_oap_count} resource OAPs and {value_oap_count} value OAPs on {host}... *******")
 
     failed_iris = _launch_thread_pool(oaps, nthreads, dsp_client)

dsp_permissions_scripts/template.py

@@ -19,10 +19,10 @@
 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 = ResourceOap(resource_iri=oap.resource_oap.resource_iri, 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