Closes #17793: Introduce a REST API endpoint for tagged objects (#18679)

* Closes #17793: Introduce a REST API endpoint for tagged objects

* Add missing object_id filter to TaggedItemFilterSet

netbox/extras/api/serializers_/tags.py

@@ -1,10 +1,16 @@
+from drf_spectacular.utils import extend_schema_field
+from rest_framework import serializers
+
 from core.models import ObjectType
-from extras.models import Tag
+from extras.models import Tag, TaggedItem
+from netbox.api.exceptions import SerializerNotFound
 from netbox.api.fields import ContentTypeField, RelatedObjectCountField
-from netbox.api.serializers import ValidatedModelSerializer
+from netbox.api.serializers import BaseModelSerializer, ValidatedModelSerializer
+from utilities.api import get_serializer_for_model
 
 __all__ = (
     'TagSerializer',
+    'TaggedItemSerializer',
 )
 
 
@@ -25,3 +31,37 @@ class TagSerializer(ValidatedModelSerializer):
             'tagged_items', 'created', 'last_updated',
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'slug', 'color', 'description')
+
+
+class TaggedItemSerializer(BaseModelSerializer):
+    object_type = ContentTypeField(
+        source='content_type',
+        read_only=True
+    )
+    object = serializers.SerializerMethodField(
+        read_only=True
+    )
+    tag = TagSerializer(
+        nested=True,
+        read_only=True
+    )
+
+    class Meta:
+        model = TaggedItem
+        fields = [
+            'id', 'url', 'display', 'object_type', 'object_id', 'object', 'tag',
+        ]
+        brief_fields = ('id', 'url', 'display', 'object_type', 'object_id', 'object', 'tag')
+
+    @extend_schema_field(serializers.JSONField())
+    def get_object(self, obj):
+        """
+        Serialize a nested representation of the tagged object.
+        """
+        try:
+            serializer = get_serializer_for_model(obj.content_object)
+        except SerializerNotFound:
+            return obj.object_repr
+        data = serializer(obj.content_object, nested=True, context={'request': self.context['request']}).data
+
+        return data

netbox/extras/api/urls.py

@@ -19,6 +19,7 @@ router.register('notifications', views.NotificationViewSet)
 router.register('notification-groups', views.NotificationGroupViewSet)
 router.register('subscriptions', views.SubscriptionViewSet)
 router.register('tags', views.TagViewSet)
+router.register('tagged-objects', views.TaggedItemViewSet)
 router.register('image-attachments', views.ImageAttachmentViewSet)
 router.register('journal-entries', views.JournalEntryViewSet)
 router.register('config-contexts', views.ConfigContextViewSet)

netbox/extras/api/views.py

@@ -6,6 +6,7 @@ from rest_framework import status
 from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
 from rest_framework.generics import RetrieveUpdateDestroyAPIView
+from rest_framework.mixins import ListModelMixin, RetrieveModelMixin
 from rest_framework.renderers import JSONRenderer
 from rest_framework.response import Response
 from rest_framework.routers import APIRootView
@@ -20,7 +21,7 @@ from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
 from netbox.api.features import SyncedDataMixin
 from netbox.api.metadata import ContentTypeMetadata
 from netbox.api.renderers import TextRenderer
-from netbox.api.viewsets import NetBoxModelViewSet
+from netbox.api.viewsets import BaseViewSet, NetBoxModelViewSet
 from utilities.exceptions import RQWorkerNotRunningException
 from utilities.request import copy_safe_request
 from . import serializers
@@ -172,6 +173,12 @@ class TagViewSet(NetBoxModelViewSet):
     filterset_class = filtersets.TagFilterSet
 
 
+class TaggedItemViewSet(RetrieveModelMixin, ListModelMixin, BaseViewSet):
+    queryset = TaggedItem.objects.prefetch_related('content_type', 'content_object', 'tag')
+    serializer_class = serializers.TaggedItemSerializer
+    filterset_class = filtersets.TaggedItemFilterSet
+
+
 #
 # Image attachments
 #

netbox/extras/filtersets.py

@@ -31,6 +31,7 @@ __all__ = (
     'SavedFilterFilterSet',
     'ScriptFilterSet',
     'TagFilterSet',
+    'TaggedItemFilterSet',
     'WebhookFilterSet',
 )
 
@@ -492,6 +493,41 @@ class TagFilterSet(ChangeLoggedModelFilterSet):
         )
 
 
+class TaggedItemFilterSet(BaseFilterSet):
+    q = django_filters.CharFilter(
+        method='search',
+        label=_('Search'),
+    )
+    object_type = ContentTypeFilter(
+        field_name='content_type'
+    )
+    object_type_id = django_filters.ModelMultipleChoiceFilter(
+        queryset=ContentType.objects.all(),
+        field_name='content_type_id'
+    )
+    tag_id = django_filters.ModelMultipleChoiceFilter(
+        queryset=Tag.objects.all()
+    )
+    tag = django_filters.ModelMultipleChoiceFilter(
+        field_name='tag__slug',
+        queryset=Tag.objects.all(),
+        to_field_name='slug',
+    )
+
+    class Meta:
+        model = TaggedItem
+        fields = ('id', 'object_id')
+
+    def search(self, queryset, name, value):
+        if not value.strip():
+            return queryset
+        return queryset.filter(
+            Q(tag__name__icontains=value) |
+            Q(tag__slug__icontains=value) |
+            Q(tag__description__icontains=value)
+        )
+
+
 class ConfigContextFilterSet(ChangeLoggedModelFilterSet):
     q = django_filters.CharFilter(
         method='search',

netbox/extras/models/tags.py

@@ -9,6 +9,7 @@ from netbox.choices import ColorChoices
 from netbox.models import ChangeLoggedModel
 from netbox.models.features import CloningMixin, ExportTemplatesMixin
 from utilities.fields import ColorField
+from utilities.querysets import RestrictedQuerySet
 
 __all__ = (
     'Tag',
@@ -72,6 +73,7 @@ class TaggedItem(GenericTaggedItemBase):
     )
 
     _netbox_private = True
+    objects = RestrictedQuerySet.as_manager()
 
     class Meta:
         indexes = [models.Index(fields=["content_type", "object_id"])]

netbox/extras/tests/test_api.py

@@ -538,6 +538,34 @@ class TagTest(APIViewTestCases.APIViewTestCase):
         Tag.objects.bulk_create(tags)
 
 
+class TaggedItemTest(
+    APIViewTestCases.GetObjectViewTestCase,
+    APIViewTestCases.ListObjectsViewTestCase
+):
+    model = TaggedItem
+    brief_fields = ['display', 'id', 'object', 'object_id', 'object_type', 'tag', 'url']
+
+    @classmethod
+    def setUpTestData(cls):
+
+        tags = (
+            Tag(name='Tag 1', slug='tag-1'),
+            Tag(name='Tag 2', slug='tag-2'),
+            Tag(name='Tag 3', slug='tag-3'),
+        )
+        Tag.objects.bulk_create(tags)
+
+        sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(sites)
+        sites[0].tags.set([tags[0], tags[1]])
+        sites[1].tags.set([tags[1], tags[2]])
+        sites[2].tags.set([tags[2], tags[0]])
+
+
 # TODO: Standardize to APIViewTestCase (needs create & update tests)
 class ImageAttachmentTest(
     APIViewTestCases.GetObjectViewTestCase,

netbox/extras/tests/test_filtersets.py

@@ -1250,6 +1250,62 @@ class TagTestCase(TestCase, ChangeLoggedFilterSetTests):
         )
 
 
+class TaggedItemFilterSetTestCase(TestCase):
+    queryset = TaggedItem.objects.all()
+    filterset = TaggedItemFilterSet
+
+    @classmethod
+    def setUpTestData(cls):
+        tags = (
+            Tag(name='Tag 1', slug='tag-1'),
+            Tag(name='Tag 2', slug='tag-2'),
+            Tag(name='Tag 3', slug='tag-3'),
+        )
+        Tag.objects.bulk_create(tags)
+
+        sites = (
+            Site(name='Site 1', slug='site-1'),
+            Site(name='Site 2', slug='site-2'),
+            Site(name='Site 3', slug='site-3'),
+        )
+        Site.objects.bulk_create(sites)
+        sites[0].tags.add(tags[0])
+        sites[1].tags.add(tags[1])
+        sites[2].tags.add(tags[2])
+
+        tenants = (
+            Tenant(name='Tenant 1', slug='tenant-1'),
+            Tenant(name='Tenant 2', slug='tenant-2'),
+            Tenant(name='Tenant 3', slug='tenant-3'),
+        )
+        Tenant.objects.bulk_create(tenants)
+        tenants[0].tags.add(tags[0])
+        tenants[1].tags.add(tags[1])
+        tenants[2].tags.add(tags[2])
+
+    def test_tag(self):
+        tags = Tag.objects.all()[:2]
+        params = {'tag': [tags[0].slug, tags[1].slug]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+        params = {'tag_id': [tags[0].pk, tags[1].pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+
+    def test_object_type(self):
+        object_type = ObjectType.objects.get_for_model(Site)
+        params = {'object_type': 'dcim.site'}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        params = {'object_type_id': [object_type.pk]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+
+    def test_object_id(self):
+        site_ids = Site.objects.values_list('pk', flat=True)
+        params = {
+            'object_type': 'dcim.site',
+            'object_id': site_ids[:2],
+        }
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+
 class ChangeLoggedFilterSetTestCase(TestCase):
     """
     Evaluate base ChangeLoggedFilterSet filters using the Site model.