make Oap.resource_oap mandatory

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

@@ -102,11 +102,10 @@ def _get_next_page(
 
 
 def _get_oap_of_one_resource(r: dict[str, Any], oap_config: OapRetrieveConfig) -> Oap | None:
-    if oap_config.retrieve_resources == "all" or r["@type"] in oap_config.specified_res_classes:
-        scope = create_scope_from_string(r["knora-api:hasPermissions"])
-        resource_oap = ResourceOap(scope=scope, resource_iri=r["@id"])
-    else:
-        resource_oap = None
+    if oap_config.retrieve_resources != "all" and r["@type"] not in oap_config.specified_res_classes:
+        return None
+    scope = create_scope_from_string(r["knora-api:hasPermissions"])
+    resource_oap = ResourceOap(scope=scope, resource_iri=r["@id"])
 
     if oap_config.retrieve_values == "none":
         value_oaps = []
@@ -115,10 +114,7 @@ def _get_oap_of_one_resource(r: dict[str, Any], oap_config: OapRetrieveConfig) -
     else:
         value_oaps = _get_value_oaps(r, oap_config.specified_props)
 
-    if resource_oap or value_oaps:
-        return Oap(resource_oap=resource_oap, value_oaps=value_oaps)
-    else:
-        return None
+    return Oap(resource_oap=resource_oap, value_oaps=value_oaps)
 
 
 def _get_value_oaps(resource: dict[str, Any], restrict_to_props: list[str] | None = None) -> list[ValueOap]:

dsp_permissions_scripts/oap/oap_model.py

@@ -6,7 +6,6 @@
 from pydantic import ConfigDict
 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
@@ -18,18 +17,11 @@ class Oap(BaseModel):
     """
     Model representing an object access permission of a resource and its values.
     If only the resource is of interest, value_oaps will be an empty list.
-    If only the values (or a part of them) are of interest, resource_oap will be None.
     """
 
-    resource_oap: ResourceOap | None
+    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 ResourceOap(BaseModel):
     """Model representing an object access permission of a resource"""

dsp_permissions_scripts/oap/oap_serialize.py

@@ -27,18 +27,12 @@ def serialize_oaps(
     folder = _get_project_data_path(shortcode, mode)
     folder.mkdir(parents=True, exist_ok=True)
     value_oap_count = sum(len(oap.value_oaps) for oap in oaps)
-    res_oap_count = sum(1 for oap in oaps if oap.resource_oap)
-    counting_info = f"{len(oaps)} OAPs ({res_oap_count} resource OAPs and {value_oap_count} value OAPs)"
-    logger.info(f"Writing {counting_info} into {folder}")
-    counter = 0
+    logger.info(f"Writing {len(oaps)} resource OAPs and {value_oap_count} value OAPs into {folder}")
     for oap in oaps:
-        if oap.resource_oap:
-            _serialize_oap(oap.resource_oap, folder)
-            counter += 1
+        _serialize_oap(oap.resource_oap, folder)
         for value_oap in oap.value_oaps:
             _serialize_oap(value_oap, folder)
-            counter += 1
-    logger.info(f"Successfully wrote {counting_info} into {counter} files in folder {folder}")
+    logger.info(f"Successfully wrote {len(oaps)} resource OAPs and {value_oap_count} value OAPs into folder {folder}")
 
 
 def _serialize_oap(oap: ResourceOap | ValueOap, folder: Path) -> None:
@@ -87,14 +81,13 @@ def sort_algo(x: ValueOap) -> str:
         return x.resource_iri
 
     for res_iri, _val_oaps in itertools.groupby(sorted(val_oaps, key=sort_algo), key=sort_algo):
-        res_oaps_filtered = [x for x in res_oaps if x.resource_iri == res_iri]
-        res_oap = res_oaps_filtered[0] if res_oaps_filtered else None
+        res_oap = next(x for x in res_oaps if x.resource_iri == res_iri)
         oaps.append(Oap(resource_oap=res_oap, value_oaps=sorted(_val_oaps, key=lambda x: x.value_iri)))
         deserialized_resource_iris.append(res_iri)
 
     remaining_res_oaps = [oap for oap in res_oaps if oap.resource_iri not in deserialized_resource_iris]
     oaps.extend(Oap(resource_oap=res_oap, value_oaps=[]) for res_oap in remaining_res_oaps)
 
-    oaps.sort(key=lambda oap: oap.resource_oap.resource_iri if oap.resource_oap else "")
+    oaps.sort(key=lambda oap: oap.resource_oap.resource_iri)
     logger.debug(f"Grouped {len(res_oaps)} resource OAPs and {len(val_oaps)} value OAPs into {len(oaps)} OAPs")
     return oaps

dsp_permissions_scripts/template.py

@@ -53,10 +53,9 @@ def modify_oaps(oaps: list[Oap]) -> list[ResourceOap | ValueOap]:
     """Adapt this sample to your needs."""
     modified_oaps: list[ResourceOap | ValueOap] = []
     for oap in copy.deepcopy(oaps):
-        if oap.resource_oap:
-            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)
+        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)
         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)

tests/test_oap_model.py

@@ -1,7 +1,6 @@
 import pytest
 
 from dsp_permissions_scripts.models import group
-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
@@ -48,26 +47,6 @@ def test_oap_multiple_vals(self) -> None:
         assert oap.resource_oap == res_oap
         assert oap.value_oaps == [val_oap_1, val_oap_2]
 
-    def test_oap_no_res(self) -> None:
-        res_iri = "http://rdfh.ch/0803/foo"
-        scope = PermissionScope.create(D=[group.UNKNOWN_USER])
-        val_oaps = [
-            ValueOap(
-                scope=scope,
-                property="foo:prop",
-                value_type="foo:valtype",
-                value_iri=f"{res_iri}/values/bar",
-                resource_iri=res_iri,
-            )
-        ]
-        oap = Oap(resource_oap=None, value_oaps=val_oaps)
-        assert oap.resource_oap is None
-        assert oap.value_oaps == val_oaps
-
-    def test_oap_no_res_no_vals(self) -> None:
-        with pytest.raises(OapEmptyError):
-            Oap(resource_oap=None, value_oaps=[])
-
 
 class TestOapRetrieveConfig:
     def test_all_resources_all_values(self) -> None:

tests/test_oap_serialization.py

@@ -3,6 +3,7 @@
 from typing import Iterator
 
 import pytest
+from pytest_unordered import unordered
 
 from dsp_permissions_scripts.models import group
 from dsp_permissions_scripts.models.scope import PermissionScope
@@ -24,15 +25,15 @@ def _setup_teardown(self) -> Iterator[None]:
             shutil.rmtree(testdata_dir)
 
     def test_oap_serialization(self) -> None:
-        oap1 = self._get_oap_one_value_only()
-        oap2 = self._get_oap_full()
+        oap1 = self._get_oap_one_value()
+        oap2 = self._get_oap_multiple_values()
         oap3 = self._get_oap_res_only()
         oaps_original = [oap1, oap2, oap3]
         serialize_oaps(oaps_original, self.shortcode, "original")
         deserialized_oaps = deserialize_oaps(self.shortcode, "original")
-        assert oaps_original == deserialized_oaps
+        assert unordered(oaps_original) == deserialized_oaps
 
-    def _get_oap_full(self) -> Oap:
+    def _get_oap_multiple_values(self) -> Oap:
         scope = PermissionScope.create(CR=[group.PROJECT_ADMIN], V=[group.PROJECT_MEMBER])
         res_iri = f"http://rdfh.ch/{self.shortcode}/resource-1"
         res_oap = ResourceOap(scope=scope, resource_iri=res_iri)
@@ -53,17 +54,18 @@ def _get_oap_full(self) -> Oap:
         oap = Oap(resource_oap=res_oap, value_oaps=[val1_oap, val2_oap])
         return oap
 
-    def _get_oap_one_value_only(self) -> Oap:
+    def _get_oap_one_value(self) -> Oap:
         scope = PermissionScope.create(D=[group.SYSTEM_ADMIN], M=[group.KNOWN_USER])
         res_iri = f"http://rdfh.ch/{self.shortcode}/resource-2"
+        res_oap = ResourceOap(scope=scope, resource_iri=res_iri)
         val_oap = ValueOap(
             scope=scope,
             property="foo:prop3",
             value_type="bar:val3",
             value_iri=f"{res_iri}/values/foobar3",
             resource_iri=res_iri,
         )
-        return Oap(resource_oap=None, value_oaps=[val_oap])
+        return Oap(resource_oap=res_oap, value_oaps=[val_oap])
 
     def _get_oap_res_only(self) -> Oap:
         scope = PermissionScope.create(V=[group.KNOWN_USER], RV=[group.UNKNOWN_USER])