class DatasetAccessPolicy(BaseAccessPolicy):
statements = [
{
"action": [
"update",
"destroy",
"partial_update",
"new_version",
"create_draft",
"publish",
],
"principal": "authenticated",
"effect": "allow",
"condition": "is_edit_allowed",
},
{
"action": "convert_from_legacy",
"principal": "*",
"effect": "allow",
},
{
# Note that there is no actual "download" action in the viewset at the moment.
"action": ["<op:download>"],
"principal": "*",
"effect": "allow",
"condition": "is_download_allowed",
},
{
"action": ["<op:custom_metadata_owner>"],
"principal": "group:service",
"effect": "allow",
},
{
"action": ["<op:flush>"], # hard delete dataset
"effect": "allow",
"principal": "authenticated",
"condition": ["is_edit_allowed", "is_flush_allowed"],
},
{
"action": ["create"],
"principal": "authenticated",
"effect": "allow", # Catalog permission checked in viewset perform_create
},
] + BaseAccessPolicy.statements
def is_edit_allowed(self, request, view, action) -> bool:
dataset = view.get_object()
return dataset.has_permission_to_edit(request.user)
def is_flush_allowed(self, request, view, action) -> bool:
dataset = view.get_object()
if dataset.state == "draft":
return True # Drafts are always hard deleted
if catalog := dataset.data_catalog:
# Hard delete is allowed for catalog admins in harvested catalogs
return catalog.harvested and DataCatalogAccessPolicy().query_object_permission(
user=request.user, object=catalog, action="<op:admin_dataset>"
)
return False
def is_metadata_owner(self, request, view, action) -> bool:
dataset = view.get_object()
logger.info(request.user)
if dataset.metadata_owner:
return request.user == dataset.metadata_owner.user
else:
return False
def is_download_allowed(self, request, view, action) -> bool:
dataset = view.get_object()
if access_rights := dataset.access_rights:
return access_rights.is_data_available(request, dataset)
return False
@classmethod
def scope_queryset(cls, request, queryset):
from .models import Dataset
if (q := super().scope_queryset(request, queryset)) is not None:
return q
elif request.user.is_anonymous:
return queryset.filter(state=Dataset.StateChoices.PUBLISHED)
groups = request.user.groups.all()
return queryset.filter(
Q(state=Dataset.StateChoices.PUBLISHED)
| Q(metadata_owner__user=request.user)
| Q(system_creator=request.user)
| Q(data_catalog__dataset_groups_admin__in=groups)
).distinct()
@classmethod
def scope_queryset_owned_or_shared(cls, request, queryset):
from .models import Dataset
if request.user.is_anonymous:
return Dataset.available_objects.none()
return queryset.filter(metadata_owner__user=request.user)