From 636939bfb7e1a8700f3aab5ba3b12113917b53ac Mon Sep 17 00:00:00 2001 From: Johannes Nussbaum Date: Mon, 15 Apr 2024 21:21:43 +0200 Subject: [PATCH] black --- dsp_permissions_scripts/oap/oap_model.py | 2 ++ dsp_permissions_scripts/oap/oap_serialize.py | 12 ++++--- tests/test_oap_serialization.py | 34 ++++++++++---------- 3 files changed, 26 insertions(+), 22 deletions(-) diff --git a/dsp_permissions_scripts/oap/oap_model.py b/dsp_permissions_scripts/oap/oap_model.py index e474909f..375a29fe 100644 --- a/dsp_permissions_scripts/oap/oap_model.py +++ b/dsp_permissions_scripts/oap/oap_model.py @@ -15,12 +15,14 @@ class Oap(BaseModel): 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] class ResourceOap(BaseModel): """Model representing an object access permission of a resource""" + scope: PermissionScope resource_iri: str diff --git a/dsp_permissions_scripts/oap/oap_serialize.py b/dsp_permissions_scripts/oap/oap_serialize.py index a3c56990..8a7e9c4c 100644 --- a/dsp_permissions_scripts/oap/oap_serialize.py +++ b/dsp_permissions_scripts/oap/oap_serialize.py @@ -56,7 +56,9 @@ def deserialize_oaps( return oaps -def _read_all_oaps_from_files(shortcode: str, mode: Literal["original", "modified"]) -> tuple[list[ResourceOap], list[ValueOap]]: +def _read_all_oaps_from_files( + shortcode: str, mode: Literal["original", "modified"] +) -> tuple[list[ResourceOap], list[ValueOap]]: folder = _get_project_data_path(shortcode, mode) res_oaps: list[ResourceOap] = [] val_oaps: list[ValueOap] = [] @@ -68,15 +70,15 @@ def _read_all_oaps_from_files(shortcode: str, mode: Literal["original", "modifie 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]: def _iri_filter(res_oap: ResourceOap, iri: str) -> bool: return res_oap.resource_iri == iri - + oaps: list[Oap] = [] deserialized_resource_iris = [] - + for res_iri, val_oaps in itertools.groupby(val_oaps, key=lambda x: x.resource_iri): filtered = list(filter(_iri_filter, res_oaps)) res_oap = filtered[0] if filtered else None @@ -86,6 +88,6 @@ def _iri_filter(res_oap: ResourceOap, iri: str) -> bool: 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) return oaps diff --git a/tests/test_oap_serialization.py b/tests/test_oap_serialization.py index a26698ce..14f1878d 100644 --- a/tests/test_oap_serialization.py +++ b/tests/test_oap_serialization.py @@ -35,34 +35,34 @@ def _get_oap_full(self) -> Oap: res_iri = f"http://rdfh.ch/{self.shortcode}/resource-1" res_oap = ResourceOap(scope=scope, resource_iri=res_iri) val1_oap = ValueOap( - scope=scope, - property="foo:prop1", - value_type="bar:val1", - value_iri=f"{res_iri}/values/foobar1", - resource_iri=res_iri + scope=scope, + property="foo:prop1", + value_type="bar:val1", + value_iri=f"{res_iri}/values/foobar1", + resource_iri=res_iri, ) val2_oap = ValueOap( - scope=scope, - property="foo:prop2", - value_type="bar:val2", - value_iri=f"{res_iri}/values/foobar2", - resource_iri=res_iri + scope=scope, + property="foo:prop2", + value_type="bar:val2", + value_iri=f"{res_iri}/values/foobar2", + resource_iri=res_iri, ) oap = Oap(resource_oap=res_oap, value_oaps=[val1_oap, val2_oap]) return oap - + def _get_oap_one_value_only(self) -> Oap: scope = PermissionScope.create(D=[group.SYSTEM_ADMIN], M=[group.KNOWN_USER]) res_iri = f"http://rdfh.ch/{self.shortcode}/resource-2" val_oap = ValueOap( - scope=scope, - property="foo:prop3", - value_type="bar:val3", - value_iri=f"{res_iri}/values/foobar3", - resource_iri=res_iri + 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]) - + def _get_oap_res_only(self) -> Oap: scope = PermissionScope.create(V=[group.KNOWN_USER], RV=[group.UNKNOWN_USER]) res_iri = f"http://rdfh.ch/{self.shortcode}/resource-3"