Update part filters (#7264)

* Expose filter for "bom_valid" status

* Expose part filter for "starred" status

* Bump API version

* Add simple unit test

* Add unit test for "starred" filtering

src/backend/InvenTree/InvenTree/api_version.py

@@ -1,11 +1,15 @@
 """InvenTree API version information."""
 
 # InvenTree API version
-INVENTREE_API_VERSION = 198
+INVENTREE_API_VERSION = 199
 """Increment this API version number whenever there is a significant change to the API that any clients need to know about."""
 
 INVENTREE_API_TEXT = """
 
+v199 - 2024-05-20 : https://github.com/inventree/InvenTree/pull/7264
+    - Expose "bom_valid" filter for the Part API
+    - Expose "starred" filter for the Part API
+
 v198 - 2024-05-19 : https://github.com/inventree/InvenTree/pull/7258
     - Fixed lookup field conflicts in the plugins API
 

src/backend/InvenTree/part/api.py

@@ -1106,6 +1106,42 @@ def filter_depleted_stock(self, queryset, name, value):
         label='Default Location', queryset=StockLocation.objects.all()
     )
 
+    bom_valid = rest_filters.BooleanFilter(
+        label=_('BOM Valid'), method='filter_bom_valid'
+    )
+
+    def filter_bom_valid(self, queryset, name, value):
+        """Filter by whether the BOM for the part is valid or not."""
+        # Limit queryset to active assemblies
+        queryset = queryset.filter(active=True, assembly=True).distinct()
+
+        # Iterate through the queryset
+        # TODO: We should cache BOM checksums to make this process more efficient
+        pks = []
+
+        for part in queryset:
+            if part.is_bom_valid() == value:
+                pks.append(part.pk)
+
+        return queryset.filter(pk__in=pks)
+
+    starred = rest_filters.BooleanFilter(label='Starred', method='filter_starred')
+
+    def filter_starred(self, queryset, name, value):
+        """Filter by whether the Part is 'starred' by the current user."""
+        if self.request.user.is_anonymous:
+            return queryset
+
+        starred_parts = [
+            star.part.pk
+            for star in self.request.user.starred_parts.all().prefetch_related('part')
+        ]
+
+        if value:
+            return queryset.filter(pk__in=starred_parts)
+        else:
+            return queryset.exclude(pk__in=starred_parts)
+
     is_template = rest_filters.BooleanFilter()
 
     assembly = rest_filters.BooleanFilter()
@@ -1235,26 +1271,6 @@ def filter_queryset(self, queryset):
 
             queryset = queryset.exclude(pk__in=id_values)
 
-        # Filter by whether the BOM has been validated (or not)
-        bom_valid = params.get('bom_valid', None)
-
-        # TODO: Querying bom_valid status may be quite expensive
-        # TODO: (It needs to be profiled!)
-        # TODO: It might be worth caching the bom_valid status to a database column
-        if bom_valid is not None:
-            bom_valid = str2bool(bom_valid)
-
-            # Limit queryset to active assemblies
-            queryset = queryset.filter(active=True, assembly=True)
-
-            pks = []
-
-            for prt in queryset:
-                if prt.is_bom_valid() == bom_valid:
-                    pks.append(prt.pk)
-
-            queryset = queryset.filter(pk__in=pks)
-
         # Filter by 'related' parts?
         related = params.get('related', None)
         exclude_related = params.get('exclude_related', None)
@@ -1288,20 +1304,6 @@ def filter_queryset(self, queryset):
             except (ValueError, Part.DoesNotExist):
                 pass
 
-        # Filter by 'starred' parts?
-        starred = params.get('starred', None)
-
-        if starred is not None:
-            starred = str2bool(starred)
-            starred_parts = [
-                star.part.pk for star in self.request.user.starred_parts.all()
-            ]
-
-            if starred:
-                queryset = queryset.filter(pk__in=starred_parts)
-            else:
-                queryset = queryset.exclude(pk__in=starred_parts)
-
         # Cascade? (Default = True)
         cascade = str2bool(params.get('cascade', True))
 

src/backend/InvenTree/part/models.py

@@ -281,7 +281,7 @@ def is_starred_by(self, user, **kwargs):
         """Returns True if the specified user subscribes to this category."""
         return user in self.get_subscribers(**kwargs)
 
-    def set_starred(self, user, status):
+    def set_starred(self, user, status: bool) -> None:
         """Set the "subscription" status of this PartCategory against the specified user."""
         if not user:
             return

src/backend/InvenTree/part/test_api.py

@@ -747,6 +747,50 @@ def test_filter_by_related(self):
         response = self.get(url, {'related': 1}, expected_code=200)
         self.assertEqual(len(response.data), 2)
 
+    def test_filter_by_bom_valid(self):
+        """Test the 'bom_valid' Part API filter."""
+        url = reverse('api-part-list')
+
+        n = Part.objects.filter(active=True, assembly=True).count()
+
+        # Initially, there are no parts with a valid BOM
+        response = self.get(url, {'bom_valid': False}, expected_code=200)
+        n1 = len(response.data)
+
+        for item in response.data:
+            self.assertTrue(item['assembly'])
+            self.assertTrue(item['active'])
+
+        response = self.get(url, {'bom_valid': True}, expected_code=200)
+        n2 = len(response.data)
+
+        self.assertEqual(n1 + n2, n)
+
+    def test_filter_by_starred(self):
+        """Test by 'starred' filter."""
+        url = reverse('api-part-list')
+
+        # All parts
+        n = Part.objects.count()
+
+        # Initially, there are no starred parts
+        response = self.get(url, {'starred': True}, expected_code=200)
+        self.assertEqual(len(response.data), 0)
+
+        response = self.get(url, {'starred': False, 'limit': 1}, expected_code=200)
+        self.assertEqual(response.data['count'], n)
+
+        # Star a part
+        part = Part.objects.first()
+        part.set_starred(self.user, True)
+
+        # Fetch data again
+        response = self.get(url, {'starred': True}, expected_code=200)
+        self.assertEqual(len(response.data), 1)
+
+        response = self.get(url, {'starred': False, 'limit': 1}, expected_code=200)
+        self.assertEqual(response.data['count'], n - 1)
+
     def test_filter_by_convert(self):
         """Test that we can correctly filter the Part list by conversion options."""
         category = PartCategory.objects.get(pk=3)