feat: implement tagging rest api tag object [FC-0030]

openedx_learning/__init__.py

@@ -1,4 +1,4 @@
 """
 Open edX Learning ("Learning Core").
 """
-__version__ = "0.1.5"
+__version__ = "0.1.6"

openedx_tagging/core/tagging/models/base.py

@@ -385,11 +385,30 @@ def tag_object(
         """
         Replaces the existing ObjectTag entries for the current taxonomy + object_id with the given list of tags.
         If self.allows_free_text, then the list should be a list of tag values.
-        Otherwise, it should be a list of existing Tag IDs.
+        Otherwise, it should be either a list of existing Tag Values or IDs.
         Raised ValueError if the proposed tags are invalid for this taxonomy.
         Preserves existing (valid) tags, adds new (valid) tags, and removes omitted (or invalid) tags.
         """
 
+        def _find_object_tag_index(tag_ref, object_tags) -> int:
+            """
+            Search for Tag in the given list of ObjectTags by tag_ref or value,
+            returning its index or -1 if not found.
+            """
+            return next(
+                (
+                    i
+                    for i, object_tag in enumerate(object_tags)
+                    if object_tag.tag_ref == tag_ref or object_tag.value == tag_ref
+                ),
+                -1,
+            )
+
+        if not isinstance(tags, list):
+            raise ValueError(_(f"Tags must be a list, not {type(tags).__name__}."))
+
+        tags = list(dict.fromkeys(tags))  # Remove duplicates preserving order
+
         if not self.allow_multiple and len(tags) > 1:
             raise ValueError(_(f"Taxonomy ({self.id}) only allows one tag per object."))
 
@@ -399,17 +418,17 @@ def tag_object(
             )
 
         ObjectTagClass = self.object_tag_class
-        current_tags = {
-            tag.tag_ref: tag
-            for tag in ObjectTagClass.objects.filter(
+        current_tags = list(
+            ObjectTagClass.objects.filter(
                 taxonomy=self,
                 object_id=object_id,
             )
-        }
+        )
         updated_tags = []
         for tag_ref in tags:
-            if tag_ref in current_tags:
-                object_tag = current_tags.pop(tag_ref)
+            object_tag_index = _find_object_tag_index(tag_ref, current_tags)
+            if object_tag_index >= 0:
+                object_tag = current_tags.pop(object_tag_index)
             else:
                 object_tag = ObjectTagClass(
                     taxonomy=self,
@@ -429,7 +448,7 @@ def tag_object(
             object_tag.save()
 
         # ...and delete any omitted existing tags
-        for old_tag in current_tags.values():
+        for old_tag in current_tags:
             old_tag.delete()
 
         return updated_tags

openedx_tagging/core/tagging/rest_api/v1/serializers.py

@@ -55,3 +55,19 @@ class Meta:
             "tag_ref",
             "is_valid",
         ]
+
+
+class ObjectTagUpdateBodySerializer(serializers.Serializer):
+    """
+    Serializer of the body for the ObjectTag UPDATE view
+    """
+
+    tags = serializers.ListField(child=serializers.CharField(), required=True)
+
+
+class ObjectTagUpdateQueryParamsSerializer(serializers.Serializer):
+    """
+    Serializer of the query params for the ObjectTag UPDATE view
+    """
+
+    taxonomy = serializers.PrimaryKeyRelatedField(queryset=Taxonomy.objects.all(), required=True)

openedx_tagging/core/tagging/rest_api/v1/views.py

@@ -3,17 +3,19 @@
 """
 from django.db import models
 from django.http import Http404
-from django.shortcuts import get_object_or_404
-from rest_framework import status
-from rest_framework.response import Response
-from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet
+from rest_framework import mixins
+from rest_framework.exceptions import MethodNotAllowed, PermissionDenied, ValidationError
+from rest_framework.viewsets import GenericViewSet, ModelViewSet
 
-from ...api import create_taxonomy, get_object_tags, get_taxonomies, get_taxonomy
+from ...api import create_taxonomy, get_object_tags, get_taxonomies, get_taxonomy, tag_object
 from ...models import Taxonomy
+from ...rules import ChangeObjectTagPermissionItem
 from .permissions import ObjectTagObjectPermissions, TaxonomyObjectPermissions
 from .serializers import (
     ObjectTagListQueryParamsSerializer,
     ObjectTagSerializer,
+    ObjectTagUpdateBodySerializer,
+    ObjectTagUpdateQueryParamsSerializer,
     TaxonomyListQueryParamsSerializer,
     TaxonomySerializer,
 )
@@ -167,10 +169,9 @@ def perform_create(self, serializer) -> None:
         serializer.instance = create_taxonomy(**serializer.validated_data)
 
 
-class ObjectTagView(ReadOnlyModelViewSet):
+class ObjectTagView(mixins.RetrieveModelMixin, mixins.UpdateModelMixin, mixins.ListModelMixin, GenericViewSet):
     """
-    View to retrieve paginated ObjectTags for an Object, given its Object ID.
-    (What tags does this object have?)
+    View to retrieve paginated ObjectTags for a provided Object ID (object_id).
 
     **Retrieve Parameters**
         * object_id (required): - The Object ID to retrieve ObjectTags for.
@@ -195,7 +196,14 @@ class ObjectTagView(ReadOnlyModelViewSet):
         * 403 - Permission denied
         * 405 - Method not allowed
 
+    **Update Parameters**
+        * object_id (required): - The Object ID to add ObjectTags for.
+
+    **Update Request Body**
+        * tags: List of tags to be applied to a object id. Must be a list of Tag ids or Tag values.
+
     **Update Query Returns**
+        * 200 - Success
         * 403 - Permission denied
         * 405 - Method not allowed
 
@@ -224,8 +232,8 @@ def get_queryset(self) -> models.QuerySet:
 
     def retrieve(self, request, object_id=None):
         """
-        Retrieve ObjectTags that belong to a given Object given its
-        object_id and return paginated results.
+        Retrieve ObjectTags that belong to a given object_id and
+        return paginated results.
 
         Note: We override `retrieve` here instead of `list` because we are
         passing in the Object ID (object_id) in the path (as opposed to passing
@@ -238,3 +246,58 @@ def retrieve(self, request, object_id=None):
         paginated_object_tags = self.paginate_queryset(object_tags)
         serializer = ObjectTagSerializer(paginated_object_tags, many=True)
         return self.get_paginated_response(serializer.data)
+
+    def update(self, request, object_id, partial=False):
+        """
+        Update ObjectTags that belong to a given object_id and
+        return the list of these ObjecTags paginated.
+
+        Pass a list of Tag ids or Tag values to be applied to an object id in the
+        body `tag` parameter. Passing an empty list will remove all tags from
+        the object id.
+
+        **Example Body Requests**
+
+        PUT api/tagging/v1/object_tags/:object_id
+
+        **Example Body Requests**
+        ```json
+        {
+            "tags": [1, 2, 3]
+        },
+        {
+            "tags": ["Tag 1", "Tag 2"]
+        },
+        {
+            "tags": []
+        }
+        """
+
+        if partial:
+            raise MethodNotAllowed("PATCH", detail="PATCH not allowed")
+
+        query_params = ObjectTagUpdateQueryParamsSerializer(data=request.query_params.dict())
+        query_params.is_valid(raise_exception=True)
+        taxonomy = query_params.validated_data.get("taxonomy", None)
+        taxonomy = taxonomy.cast()
+
+        perm = f"{taxonomy._meta.app_label}.change_objecttag"
+
+        perm_obj = ChangeObjectTagPermissionItem(
+            taxonomy=taxonomy,
+            object_id=object_id,
+        )
+
+        if not request.user.has_perm(perm, perm_obj):
+            raise PermissionDenied("You do not have permission to change object tags for this taxonomy or object_id.")
+
+        body = ObjectTagUpdateBodySerializer(data=request.data)
+        body.is_valid(raise_exception=True)
+
+        tags = body.data.get("tags", [])
+        try:
+            tag_object(taxonomy, tags, object_id)
+        except ValueError as e:
+            raise ValidationError(e)
+
+        return self.retrieve(request, object_id)

openedx_tagging/core/tagging/rules.py

@@ -8,8 +8,9 @@
 import django.contrib.auth.models
 # typing support in rules depends on https://github.com/dfunckt/django-rules/pull/177
 import rules  # type: ignore[import]
+from attrs import define
 
-from .models import ObjectTag, Tag, Taxonomy
+from .models import Tag, Taxonomy
 
 UserType = Union[django.contrib.auth.models.User, django.contrib.auth.models.AnonymousUser]
 
@@ -19,6 +20,16 @@
 is_taxonomy_admin: Callable[[UserType], bool] = rules.is_staff
 
 
+@define
+class ChangeObjectTagPermissionItem:
+    """
+    Pair of taxonomy and object_id used for permission checking.
+    """
+
+    taxonomy: Taxonomy
+    object_id: str
+
+
 @rules.predicate
 def can_view_taxonomy(user: UserType, taxonomy: Taxonomy | None = None) -> bool:
     """
@@ -53,18 +64,39 @@ def can_change_tag(user: UserType, tag: Tag | None = None) -> bool:
 
 
 @rules.predicate
-def can_change_object_tag(user: UserType, object_tag: ObjectTag | None = None) -> bool:
+def can_change_object_tag_objectid(_user: UserType, _object_id: str) -> bool:
     """
-    Taxonomy admins can create or modify object tags on enabled taxonomies.
+    Nobody can create or modify object tags without checking the permission for the tagged object.
+
+    This rule should be defined in other apps for proper permission checking.
     """
-    taxonomy = (
-        object_tag.taxonomy.cast() if (object_tag and object_tag.taxonomy) else None
-    )
-    object_tag = taxonomy.object_tag_class.cast(object_tag) if taxonomy else object_tag
-    return is_taxonomy_admin(user) and (
-        not object_tag or not taxonomy or (taxonomy and taxonomy.enabled)
+    return False
+
+
+@rules.predicate
+def can_change_object_tag(user: UserType, perm_obj: ChangeObjectTagPermissionItem | None = None) -> bool:
+    """
+    Checks if the user has permissions to create or modify tags on the given taxonomy and object_id.
+    """
+
+    # The following code allows METHOD permission (PUT) in the viewset for everyone
+    if perm_obj is None:
+        return True
+
+    # Checks the permission for the taxonomy
+    taxonomy_perm = user.has_perm("oel_tagging.change_objecttag_taxonomy", perm_obj.taxonomy)
+    if not taxonomy_perm:
+        return False
+
+    # Checks the permission for the object_id
+    objectid_perm = user.has_perm(
+        "oel_tagging.change_objecttag_objectid",
+        # The obj arg expects an object, but we are passing a string
+        perm_obj.object_id,  # type: ignore[arg-type]
     )
 
+    return objectid_perm
+
 
 # Taxonomy
 rules.add_perm("oel_tagging.add_taxonomy", can_change_taxonomy)
@@ -81,5 +113,9 @@ def can_change_object_tag(user: UserType, object_tag: ObjectTag | None = None) -
 # ObjectTag
 rules.add_perm("oel_tagging.add_objecttag", can_change_object_tag)
 rules.add_perm("oel_tagging.change_objecttag", can_change_object_tag)
-rules.add_perm("oel_tagging.delete_objecttag", is_taxonomy_admin)
+rules.add_perm("oel_tagging.delete_objecttag", can_change_object_tag)
 rules.add_perm("oel_tagging.view_objecttag", rules.always_allow)
+
+# Users can tag objects using tags from any taxonomy that they have permission to view
+rules.add_perm("oel_tagging.change_objecttag_taxonomy", can_view_taxonomy)
+rules.add_perm("oel_tagging.change_objecttag_objectid", can_change_object_tag_objectid)