refactor: make object tag permission checks modular

src/openedx_core/__init__.py

@@ -6,4 +6,4 @@
 """
 
 # The version for the entire repository
-__version__ = "0.44.0"
+__version__ = "0.45.0"

src/openedx_tagging/rest_api/v1/serializers.py

@@ -145,7 +145,17 @@ class Meta:
 
     def get_can_delete_objecttag(self, instance) -> bool | None:
         """
-        Returns True if the current request user may delete object tags on this taxonomy
+        SerializerMethodField callback for `can_delete_objecttag`.
+
+        Delegates to `can_delete_object_tag` so subclasses can
+        override a stable method name that isn't tied to DRF's
+        SerializerMethodField naming convention.
+        """
+        return self.can_delete_object_tag(instance)
+
+    def can_delete_object_tag(self, instance) -> bool | None:
+        """
+        Check if the user is authorized to delete the provided tag.
         """
         perm_name = f'{self.app_label}.remove_objecttag_objectid'
         return self._can(perm_name, instance.object_id)
@@ -179,7 +189,6 @@ def to_representation(self, instance: list[ObjectTag]) -> dict:
         # Allows consumers like edx-platform to override this
         ObjectTagViewMinimalSerializer = self.context["view"].minimal_serializer_class
 
-        can_tag_object_perm = f"{self.app_label}.can_tag_object"
         by_object: dict[str, dict[str, Any]] = {}
         for obj_tag in instance:
             if obj_tag.object_id not in by_object:
@@ -192,14 +201,21 @@ def to_representation(self, instance: list[ObjectTag]) -> dict:
                 tax_entry = {
                     "name": obj_tag.taxonomy.name if obj_tag.taxonomy else None,
                     "taxonomy_id": obj_tag.taxonomy_id,
-                    "can_tag_object": self._can(can_tag_object_perm, obj_tag),
+                    "can_tag_object": self.can_tag_object(obj_tag),
                     "tags": [],
                     "export_id": obj_tag.export_id,
                 }
                 taxonomies.append(tax_entry)
             tax_entry["tags"].append(ObjectTagViewMinimalSerializer(obj_tag, context=self.context).data)
         return by_object
 
+    def can_tag_object(self, obj_tag) -> bool | None:
+        """
+        Check if the user is authorized to tag the provided object.
+        """
+        perm_name = f"{self.app_label}.can_tag_object"
+        return self._can(perm_name, obj_tag)
+
 
 class ObjectTagUpdateByTaxonomySerializer(serializers.Serializer):  # pylint: disable=abstract-method
     """

src/openedx_tagging/rest_api/v1/views.py

@@ -448,6 +448,8 @@ class ObjectTagView(
 
     # Serializer used in `get_queryset` when getting tags per taxonomy
     serializer_class = ObjectTagSerializer
+    # Serializer used in `retrieve` to represent all tags of a given object
+    object_tags_serializer_class = ObjectTagsByTaxonomySerializer
     # Serializer used in the result in `to_representation` in `ObjectTagsByTaxonomySerializer`
     minimal_serializer_class = ObjectTagMinimalSerializer
     permission_classes = [ObjectTagObjectPermissions]
@@ -471,23 +473,29 @@ def get_queryset(self) -> models.QuerySet:
             taxonomy = taxonomy.cast()
             taxonomy_id = taxonomy.id
 
-        if object_id.endswith("*") or "," in object_id:  # pylint: disable=no-else-raise
-            raise ValidationError("Retrieving tags from multiple objects is not yet supported.")
+        if object_id.endswith("*") or "," in object_id:
             # Note: This API is actually designed so that in the future it can be extended to return tags for multiple
             # objects, e.g. if object_id.endswith("*") then it results in a object_id__startswith query. However, for
             # now we have no use case for that so we retrieve tags for one object at a time.
-        else:
-            if not self.request.user.has_perm(
-                "oel_tagging.view_objecttag",
-                # The obj arg expects a model, but we are passing an object
-                ObjectTagPermissionItem(taxonomy=taxonomy, object_id=object_id),  # type: ignore[arg-type]
-            ):
-                raise PermissionDenied(
-                    "You do not have permission to view object tags for this taxonomy or object_id."
-                )
+            raise ValidationError("Retrieving tags from multiple objects is not yet supported.")
+
+        self.ensure_has_view_object_tag_permission(self.request.user, taxonomy, object_id)
 
         return get_object_tags(object_id, taxonomy_id)
 
+    def ensure_has_view_object_tag_permission(self, user, taxonomy, object_id):
+        """
+        Check if user has permission to view object tags.
+        """
+        if not user.has_perm(
+            "oel_tagging.view_objecttag",
+            # The obj arg expects a model, but we are passing an object
+            ObjectTagPermissionItem(taxonomy=taxonomy, object_id=object_id),  # type: ignore[arg-type]
+        ):
+            raise PermissionDenied(
+                "You do not have permission to view object tags for this taxonomy or object_id."
+            )
+
     def retrieve(self, request, *args, **kwargs) -> Response:
         """
         Retrieve ObjectTags that belong to a given object_id
@@ -500,7 +508,7 @@ def retrieve(self, request, *args, **kwargs) -> Response:
         behavior we want.
         """
         object_tags = self.filter_queryset(self.get_queryset())
-        serializer = ObjectTagsByTaxonomySerializer(list(object_tags), context=self.get_serializer_context())
+        serializer = self.object_tags_serializer_class(list(object_tags), context=self.get_serializer_context())
         response_data = serializer.data
         if self.kwargs["object_id"] not in response_data:
             # For consistency, the key with the object_id should always be present in the response, even if there
@@ -514,7 +522,7 @@ def update(self, request, *args, **kwargs) -> Response:
 
         Pass a list of Tag ids or Tag values to be applied to an object id in the
         body `tag` parameter, by each taxonomy. Passing an empty list will remove all tags from
-        the object id on an specific taxonomy.
+        the object id on a specific taxonomy.
 
         **Example Body Requests**
 
@@ -550,13 +558,13 @@ def update(self, request, *args, **kwargs) -> Response:
                 },
             ]
         }
+        ```
         """
         partial = kwargs.pop('partial', False)
         if partial:
             raise MethodNotAllowed("PATCH", detail="PATCH not allowed")
 
         object_id = kwargs.pop('object_id')
-        perm = "oel_tagging.can_tag_object"
         body = ObjectTagUpdateBodySerializer(data=request.data)
         body.is_valid(raise_exception=True)
 
@@ -566,14 +574,33 @@ def update(self, request, *args, **kwargs) -> Response:
             return self.retrieve(request, object_id)
 
         # Check permissions
-        for tagsData in data:
-            taxonomy = tagsData.get("taxonomy")
+        self.ensure_user_has_can_tag_object_permissions(request.user, data, object_id)
+
+        # Tag object_id per taxonomy
+        for tag_data in data:
+            taxonomy = tag_data.get("taxonomy")
+            tags = tag_data.get("tags", [])
+            try:
+                tag_object(object_id, taxonomy, tags)
+            except TagDoesNotExist as e:
+                raise ValidationError from e
+            except ValueError as e:
+                raise ValidationError from e
 
+        return self.retrieve(request, object_id)
+
+    def ensure_user_has_can_tag_object_permissions(self, user, tags_data, object_id):
+        """
+        Check if user has permission to tag object for each taxonomy.
+        """
+        perm = "oel_tagging.can_tag_object"
+        for tag_data in tags_data:
+            taxonomy = tag_data.get("taxonomy")
             perm_obj = ObjectTagPermissionItem(
                 taxonomy=taxonomy,
                 object_id=object_id,
             )
-            if not request.user.has_perm(
+            if not user.has_perm(
                 perm,
                 # The obj arg expects a model, but we are passing an object
                 perm_obj,  # type: ignore[arg-type]
@@ -583,19 +610,6 @@ def update(self, request, *args, **kwargs) -> Response:
                     for Taxonomy: {str(taxonomy)} or Object: {object_id}.
                 """)
 
-        # Tag object_id per taxonomy
-        for tagsData in data:
-            taxonomy = tagsData.get("taxonomy")
-            tags = tagsData.get("tags", [])
-            try:
-                tag_object(object_id, taxonomy, tags)
-            except TagDoesNotExist as e:
-                raise ValidationError from e
-            except ValueError as e:
-                raise ValidationError from e
-
-        return self.retrieve(request, object_id)
-
 
 @view_auth_classes
 class ObjectTagCountsView(