feat: update a certain value only (instead of all values of the resource)

dsp_permissions_scripts/models/errors.py

@@ -17,3 +17,18 @@ def __str__(self) -> str:
 @dataclass
 class PermissionsAlreadyUpToDate(Exception):
     message: str = "The submitted permissions are the same as the current ones"
+
+
+@dataclass
+class SpecifiedPropsEmptyError(ValueError):
+    message: str = "specified_props must not be empty if retrieve_values is 'specified_props'"
+
+
+@dataclass
+class SpecifiedPropsNotEmptyError(ValueError):
+    message: str = "specified_props must be empty if retrieve_values is not 'specified_props'"
+
+
+@dataclass
+class OapRetrieveConfigEmptyError(ValueError):
+    message: str = "retrieve_resources cannot be False if retrieve_values is 'none'"

dsp_permissions_scripts/oap/oap_get.py

@@ -4,6 +4,9 @@
 
 from dsp_permissions_scripts.models.errors import ApiError
 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.utils.dsp_client import DspClient
 from dsp_permissions_scripts.utils.get_logger import get_logger
 from dsp_permissions_scripts.utils.project import get_all_resource_class_iris_of_project
@@ -13,8 +16,9 @@
 logger = get_logger(__name__)
 
 
-def _get_all_oaps_of_resclass(resclass_iri: str, project_iri: str, dsp_client: DspClient) -> list[Oap]:
-    logger.info(f"Getting all resource OAPs of class {resclass_iri}...")
+def _get_all_oaps_of_resclass(
+    resclass_iri: str, project_iri: str, dsp_client: DspClient, oap_config: OapRetrieveConfig
+) -> list[Oap]:
     headers = {"X-Knora-Accept-Project": project_iri}
     all_oaps: list[Oap] = []
     page = 0
@@ -27,6 +31,7 @@ def _get_all_oaps_of_resclass(resclass_iri: str, project_iri: str, dsp_client: D
                 page=page,
                 headers=headers,
                 dsp_client=dsp_client,
+                oap_config=oap_config,
             )
             all_oaps.extend(oaps)
             page += 1
@@ -42,6 +47,7 @@ def _get_next_page(
     page: int,
     headers: dict[str, str],
     dsp_client: DspClient,
+    oap_config: OapRetrieveConfig,
 ) -> tuple[bool, list[Oap]]:
     """
     Get the resource IRIs of a resource class, one page at a time.
@@ -63,19 +69,62 @@ def _get_next_page(
     if "@graph" in result:
         oaps = []
         for r in result["@graph"]:
-            scope = create_scope_from_string(r["knora-api:hasPermissions"])
-            oaps.append(Oap(scope=scope, object_iri=r["@id"]))
+            if oap := _get_oap_of_one_resource(r, oap_config):
+                oaps.append(oap)
         return True, oaps
 
     # result contains only 1 resource: return it, then stop (there will be no more resources)
     if "@id" in result:
-        scope = create_scope_from_string(result["knora-api:hasPermissions"])
-        return False, [Oap(scope=scope, object_iri=result["@id"])]
+        oaps = []
+        if oap := _get_oap_of_one_resource(result, oap_config):
+            oaps.append(oap)
+        return False, oaps
 
     # there are no more resources
     return False, []
 
 
+def _get_oap_of_one_resource(r: dict[str, Any], oap_config: OapRetrieveConfig) -> Oap | None:
+    if oap_config.retrieve_resources:
+        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_values == "none":
+        value_oaps = []
+    elif oap_config.retrieve_values == "all":
+        value_oaps = _get_value_oaps(r)
+    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
+
+
+def _get_value_oaps(resource: dict[str, Any], restrict_to_props: list[str] | None = None) -> list[ValueOap]:
+    res = []
+    for k, v in resource.items():
+        if k in {"@id", "@type", "@context", "rdfs:label", "knora-api:DeletedValue"}:
+            continue
+        if restrict_to_props is not None and k not in restrict_to_props:
+            continue
+        match v:
+            case {
+                "@id": id_,
+                "@type": type_,
+                "knora-api:hasPermissions": perm_str,
+            } if "/values/" in id_:
+                scope = create_scope_from_string(perm_str)
+                oap = ValueOap(scope=scope, property=k, value_type=type_, value_iri=id_, resource_iri=resource["@id"])
+                res.append(oap)
+            case _:
+                continue
+    return res
+
+
 def get_resource(resource_iri: str, dsp_client: DspClient) -> dict[str, Any]:
     """Requests the resource with the given IRI from DSP-API"""
     iri = quote_plus(resource_iri, safe="")
@@ -86,15 +135,16 @@ def get_resource(resource_iri: str, dsp_client: DspClient) -> dict[str, Any]:
         raise err from None
 
 
-def get_resource_oap_by_iri(resource_iri: str, dsp_client: DspClient) -> Oap:
+def get_resource_oap_by_iri(resource_iri: str, dsp_client: DspClient) -> ResourceOap:
     resource = get_resource(resource_iri, dsp_client)
     scope = create_scope_from_string(resource["knora-api:hasPermissions"])
-    return Oap(scope=scope, object_iri=resource_iri)
+    return ResourceOap(scope=scope, resource_iri=resource_iri)
 
 
-def get_all_resource_oaps_of_project(
+def get_all_oaps_of_project(
     shortcode: str,
     dsp_client: DspClient,
+    oap_config: OapRetrieveConfig,
     excluded_class_iris: Iterable[str] = (),
 ) -> list[Oap]:
     logger.info("******* Retrieving all OAPs... *******")
@@ -103,7 +153,7 @@ def get_all_resource_oaps_of_project(
     resclass_iris = [x for x in resclass_iris if x not in excluded_class_iris]
     all_oaps = []
     for resclass_iri in resclass_iris:
-        oaps = _get_all_oaps_of_resclass(resclass_iri, project_iri, dsp_client)
+        oaps = _get_all_oaps_of_resclass(resclass_iri, project_iri, dsp_client, oap_config)
         all_oaps.extend(oaps)
     logger.info(f"Retrieved a TOTAL of {len(all_oaps)} OAPs")
     return all_oaps

dsp_permissions_scripts/oap/oap_model.py

@@ -1,10 +1,76 @@
+from __future__ import annotations
+
+from typing import Literal
+
 from pydantic import BaseModel
+from pydantic import ConfigDict
+from pydantic import model_validator
 
+from dsp_permissions_scripts.models.errors import OapRetrieveConfigEmptyError
+from dsp_permissions_scripts.models.errors import SpecifiedPropsEmptyError
+from dsp_permissions_scripts.models.errors import SpecifiedPropsNotEmptyError
 from dsp_permissions_scripts.models.scope import PermissionScope
 
 
 class Oap(BaseModel):
-    """Model representing an object access permission, containing a scope and the IRI of the resource/value"""
+    """
+    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
+    value_oaps: list[ValueOap]
+
+    @model_validator(mode="after")
+    def check_consistency(self) -> Oap:
+        if not self.resource_oap and not self.value_oaps:
+            raise ValueError("An OAP must have at least one resource_oap or one value_oap")
+        return self
+
+
+class ResourceOap(BaseModel):
+    """Model representing an object access permission of a resource"""
+
+    scope: PermissionScope
+    resource_iri: str
+
+
+class ValueOap(BaseModel):
+    """
+    Model representing an object access permission of a value.
+
+    Fields:
+    scope: permissions of this value
+    value_iri: IRI of the value
+    property: property whith which the value relates to its resource
+    value_type: type of the value, e.g. "knora-api:TextValue"
+    """
 
     scope: PermissionScope
-    object_iri: str
+    property: str
+    value_type: str
+    value_iri: str
+    resource_iri: str
+
+
+class OapRetrieveConfig(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    retrieve_resources: bool = True
+    retrieve_values: Literal["all", "specified_props", "none"] = "none"
+    specified_props: list[str] = []
+
+    @model_validator(mode="after")
+    def check_specified_props(self) -> OapRetrieveConfig:
+        if self.retrieve_values == "specified_props" and not self.specified_props:
+            raise SpecifiedPropsEmptyError()
+        if self.retrieve_values != "specified_props" and self.specified_props:
+            raise SpecifiedPropsNotEmptyError()
+        return self
+
+    @model_validator(mode="after")
+    def check_config_empty(self) -> OapRetrieveConfig:
+        if not self.retrieve_resources and self.retrieve_values == "none":
+            raise OapRetrieveConfigEmptyError()
+        return self

dsp_permissions_scripts/oap/oap_serialize.py

@@ -1,8 +1,11 @@
+import itertools
 import re
 from pathlib import Path
 from typing import Literal
 
 from dsp_permissions_scripts.oap.oap_model import Oap
+from dsp_permissions_scripts.oap.oap_model import ResourceOap
+from dsp_permissions_scripts.oap.oap_model import ValueOap
 from dsp_permissions_scripts.utils.get_logger import get_logger
 
 logger = get_logger(__name__)
@@ -24,21 +27,63 @@ def serialize_oaps(
     folder = _get_project_data_path(shortcode, mode)
     folder.mkdir(parents=True, exist_ok=True)
     logger.info(f"Writing {len(oaps)} OAPs into {folder}")
-    for res_oap in oaps:
-        filename = re.sub(r"http://rdfh\.ch/[^/]+/", "resource_", res_oap.object_iri)
-        with open(folder / f"{filename}.json", mode="w", encoding="utf-8") as f:
-            f.write(res_oap.model_dump_json(indent=2))
-    logger.info(f"Successfully wrote {len(oaps)} OAPs into {folder}")
+    counter = 0
+    for oap in oaps:
+        if oap.resource_oap:
+            _serialize_oap(oap.resource_oap, folder)
+            counter += 1
+        for value_oap in oap.value_oaps:
+            _serialize_oap(value_oap, folder)
+            counter += 1
+    logger.info(f"Successfully wrote {len(oaps)} OAPs into {counter} files in folder {folder}")
+
+
+def _serialize_oap(oap: ResourceOap | ValueOap, folder: Path) -> None:
+    iri = oap.resource_iri if isinstance(oap, ResourceOap) else oap.value_iri
+    filename = re.sub(r"http://rdfh\.ch/[^/]+/", "resource_", iri)
+    filename = re.sub(r"/", "_", filename)
+    with open(folder / f"{filename}.json", mode="w", encoding="utf-8") as f:
+        f.write(oap.model_dump_json(indent=2))
 
 
 def deserialize_oaps(
     shortcode: str,
     mode: Literal["original", "modified"],
 ) -> list[Oap]:
-    """Deserialize the resource OAPs from JSON files."""
+    """Deserialize the OAPs from JSON files."""
+    res_oaps, val_oaps = _read_all_oaps_from_files(shortcode, mode)
+    oaps = _group_oaps_together(res_oaps, val_oaps)
+    return oaps
+
+
+def _read_all_oaps_from_files(
+    shortcode: str, mode: Literal["original", "modified"]
+) -> tuple[list[ResourceOap], list[ValueOap]]:
     folder = _get_project_data_path(shortcode, mode)
-    oaps = []
-    for file in [f for f in folder.iterdir() if f.suffix == ".json"]:
-        with open(file, mode="r", encoding="utf-8") as f:
-            oaps.append(Oap.model_validate_json(f.read()))
+    res_oaps: list[ResourceOap] = []
+    val_oaps: list[ValueOap] = []
+    for file in folder.glob("**/*.json"):
+        content = file.read_text(encoding="utf-8")
+        if "_values_" in file.name:
+            val_oaps.append(ValueOap.model_validate_json(content))
+        else:
+            res_oaps.append(ResourceOap.model_validate_json(content))
+    return res_oaps, val_oaps
+
+
+def _group_oaps_together(res_oaps: list[ResourceOap], val_oaps: list[ValueOap]) -> list[Oap]:
+    oaps: list[Oap] = []
+    deserialized_resource_iris = []
+
+    for res_iri, _val_oaps in itertools.groupby(val_oaps, key=lambda x: x.resource_iri):
+        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
+        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]
+    for res_oap in remaining_res_oaps:
+        oaps.append(Oap(resource_oap=res_oap, value_oaps=[]))
+
+    oaps.sort(key=lambda oap: oap.resource_oap.resource_iri if oap.resource_oap else "")
     return oaps