Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: retrieve oaps instead of just resource iris #19

Merged
merged 12 commits into from
Oct 4, 2023
7 changes: 7 additions & 0 deletions dsp_permissions_scripts/models/permission.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,3 +41,10 @@ class DoapTargetType(Enum):
GROUP = "group"
RESOURCE_CLASS = "resource_class"
PROPERTY = "property"


class Oap(BaseModel):
"""Model representing an object access permission, containing a scope and the IRI of the resource/value"""

scope: PermissionScope
object_iri: str
19 changes: 13 additions & 6 deletions dsp_permissions_scripts/template.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,27 @@

from dsp_permissions_scripts.models.groups import BuiltinGroup
from dsp_permissions_scripts.models.host import Hosts
from dsp_permissions_scripts.models.permission import Oap
from dsp_permissions_scripts.models.scope import PUBLIC
from dsp_permissions_scripts.utils.authentication import login
from dsp_permissions_scripts.utils.permissions import (
apply_updated_oaps_on_server,
get_doaps_of_project,
print_doaps_of_project,
set_doaps_of_groups,
update_permissions_for_resources_and_values,
)
from dsp_permissions_scripts.utils.project import (
get_all_resource_iris_of_project,
get_all_resource_oaps_of_project,
get_project_iri_by_shortcode,
)


def modify_oaps(oaps: list[Oap]) -> list[Oap]:
for oap in oaps:
oap.scope.D.append(BuiltinGroup.PROJECT_MEMBER)
return oaps


def main() -> None:
"""
The main method assembles a sample call of all available high-level functions.
Expand Down Expand Up @@ -49,14 +56,14 @@ def main() -> None:
shortcode=shortcode,
token=token,
)
resource_iris = get_all_resource_iris_of_project(
resource_oaps = get_all_resource_oaps_of_project(
project_iri=project_iri,
host=host,
token=token,
)
update_permissions_for_resources_and_values(
resource_iris=resource_iris,
scope=new_scope,
resource_oaps_updated = modify_oaps(oaps=resource_oaps)
apply_updated_oaps_on_server(
resource_oaps=resource_oaps_updated,
host=host,
token=token,
)
Expand Down
16 changes: 16 additions & 0 deletions dsp_permissions_scripts/utils/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
Doap,
DoapTarget,
DoapTargetType,
Oap,
PermissionScope,
)
from dsp_permissions_scripts.models.value import ValueUpdate
Expand Down Expand Up @@ -233,6 +234,21 @@ def update_doap_scope(
return new_doap


def apply_updated_oaps_on_server(
resource_oaps: list[Oap],
host: str,
token: str,
) -> None:
"""Applies object access permissions on a DSP server."""
for resource_oap in resource_oaps:
update_permissions_for_resources_and_values(
resource_iris=[resource_oap.object_iri],
scope=resource_oap.scope,
host=host,
token=token,
)


def update_permissions_for_resources_and_values(
resource_iris: list[str],
scope: PermissionScope,
Expand Down
35 changes: 21 additions & 14 deletions dsp_permissions_scripts/utils/project.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@

import requests

from dsp_permissions_scripts.models.permission import Oap
from dsp_permissions_scripts.utils.authentication import get_protocol
from dsp_permissions_scripts.utils.scope_serialization import create_scope_from_string


def get_project_iri_by_shortcode(shortcode: str, host: str) -> str:
Expand All @@ -17,26 +19,26 @@ def get_project_iri_by_shortcode(shortcode: str, host: str) -> str:
return iri


def get_all_resource_iris_of_project(
def get_all_resource_oaps_of_project(
project_iri: str,
host: str,
token: str,
) -> list[str]:
all_resource_iris = []
) -> list[Oap]:
all_resource_oaps = []
resclass_iris = __get_all_resource_class_iris_of_project(
project_iri=project_iri,
host=host,
token=token,
)
for resclass_iri in resclass_iris:
resource_iris = __get_all_resource_iris_of_resclass(
resource_oaps = __get_all_resource_oaps_of_resclass(
host=host,
resclass_iri=resclass_iri,
project_iri=project_iri,
token=token,
)
all_resource_iris.extend(resource_iris)
return all_resource_iris
all_resource_oaps.extend(resource_oaps)
return all_resource_oaps


def __get_all_resource_class_iris_of_project(
Expand Down Expand Up @@ -97,15 +99,15 @@ def __dereference_prefix(identifier: str, context: dict[str, str]) -> str:
return context[prefix] + actual_id


def __get_all_resource_iris_of_resclass(
def __get_all_resource_oaps_of_resclass(
host: str,
resclass_iri: str,
project_iri: str,
token: str,
) -> list[str]:
) -> list[Oap]:
protocol = get_protocol(host)
headers = {"X-Knora-Accept-Project": project_iri, "Authorization": f"Bearer {token}"}
resource_iris = []
resources: list[Oap] = []
page = 0
more = True
while more:
Expand All @@ -116,9 +118,9 @@ def __get_all_resource_iris_of_resclass(
page=page,
headers=headers,
)
resource_iris.extend(iris)
resources.extend(iris)
page += 1
return resource_iris
return resources


def __get_next_page(
Expand All @@ -127,7 +129,7 @@ def __get_next_page(
resclass_iri: str,
page: int,
headers: dict[str, str],
) -> tuple[bool, list[str]]:
) -> tuple[bool, list[Oap]]:
"""
Get the resource IRIs of a resource class, one page at a time.
DSP-API returns results page-wise:
Expand All @@ -143,10 +145,15 @@ def __get_next_page(
result = response.json()
if "@graph" in result:
# result contains several resources: return them, then continue with next page
return True, [r["@id"] for r in result["@graph"]]
oaps = []
for r in result["@graph"]:
scope = create_scope_from_string(r["knora-api:hasPermissions"])
oaps.append(Oap(scope=scope, object_iri=r["@id"]))
return True, oaps
elif "@id" in result:
# result contains only 1 resource: return it, then stop (there will be no more resources)
return False, [result["@id"]]
scope = create_scope_from_string(result["knora-api:hasPermissions"])
return False, [Oap(scope=scope, object_iri=result["@id"])]
else:
# there are no more resources
return False, []