Coverage for src/backend/InvenTree/common/filters.py: 57%
127 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 17:47 +0000
1"""Custom API filters for InvenTree."""
3import re
5from django.contrib.contenttypes.models import ContentType
6from django.core.exceptions import ValidationError
7from django.db.models import (
8 Case,
9 CharField,
10 Exists,
11 FloatField,
12 Model,
13 OuterRef,
14 Q,
15 Subquery,
16 Value,
17 When,
18)
19from django.db.models.query import QuerySet
20from django.utils.translation import gettext_lazy as _
22import django_filters.rest_framework.filters as rest_filters
23from rest_framework import serializers
24from taggit.serializers import TagListSerializerField
26import InvenTree.conversion
27import InvenTree.helpers
28import InvenTree.serializers
31def determine_content_type(content_type: str | int | None) -> ContentType | None:
32 """Determine a ContentType instance from a string or integer input.
34 Arguments:
35 content_type: The content type to resolve (name or ID).
37 Returns:
38 ContentType instance if found, else None.
39 """
40 if content_type is None: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true
41 return None
43 ct = None
45 # First, try to resolve the content type via a PK value
46 try:
47 content_type_id = int(content_type)
48 ct = ContentType.objects.get_for_id(content_type_id)
49 except (ValueError, ContentType.DoesNotExist):
50 ct = None
52 if len(content_type.split('.')) == 2:
53 # Next, try to resolve the content type via app_label.model_name
54 try:
55 app_label, model = content_type.split('.')
56 ct = ContentType.objects.get(app_label=app_label, model=model)
57 except ContentType.DoesNotExist:
58 ct = None
60 else:
61 # Next, try to resolve the content type via a model name
62 ct = ContentType.objects.filter(model__iexact=content_type).first()
64 return ct
67def filter_content_type(
68 queryset, field_name: str, content_type: str | int | None, allow_null: bool = True
69):
70 """Filter a queryset by content type.
72 Arguments:
73 queryset: The queryset to filter.
74 field_name: The name of the content type field within the current model context.
75 content_type: The content type to filter by (name or ID).
76 allow_null: If True, include entries with null content type.
78 Returns:
79 Filtered queryset.
80 """
81 if content_type is None: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 return queryset
84 ct = determine_content_type(content_type)
86 if ct is None:
87 raise ValidationError(f'Invalid content type: {content_type}')
89 q = Q(**{f'{field_name}': ct})
91 if allow_null: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 q |= Q(**{f'{field_name}__isnull': True})
94 return queryset.filter(q)
97class TagsFilter(rest_filters.CharFilter):
98 """Filter which accepts a comma-separated list of tag names and returns only objects that have ALL of the specified tags.
100 Example usage in a FilterSet:
101 tags = TagsFilter(label=_('Tags'))
103 Example query:
104 ?tags=apple,banana → returns only items tagged with both 'apple' AND 'banana'
105 """
107 def __init__(self, *args, **kwargs):
108 """Initialize the filter."""
109 if 'label' not in kwargs:
110 kwargs['label'] = _('Tags')
112 super().__init__(*args, **kwargs)
114 def filter(self, qs, value):
115 """Filter queryset to items matching all provided tag names."""
116 if not value:
117 return qs
119 tag_names = [t.strip() for t in value.split(',') if t.strip()]
121 for tag in tag_names:
122 qs = qs.filter(tags__name__iexact=tag)
124 return qs.distinct()
127"""A list of valid operators for filtering part parameters."""
128PARAMETER_FILTER_OPERATORS: list[str] = ['gt', 'gte', 'lt', 'lte', 'ne', 'icontains']
131def filter_parameters_by_value(
132 queryset: QuerySet, template_id: int, value: str, func: str = ''
133) -> QuerySet:
134 """Filter the Parameter model based on the provided template and value.
136 Arguments:
137 queryset: The initial QuerySet to filter.
138 template_id: The parameter template ID to filter by.
139 value: The value to filter against.
140 func: The filtering function to apply (e.g. 'gt', 'lt', etc).
142 Returns:
143 A list of Parameter instances which match the given criteria.
145 Notes:
146 - Parts which do not have a value for the given parameter are excluded.
147 """
148 from common.models import ParameterTemplate
150 # Ensure that the provided function is valid
151 if func and func not in PARAMETER_FILTER_OPERATORS:
152 raise ValueError(f'Invalid parameter filter function: {func}')
154 # Ensure that the template exists
155 try:
156 template = ParameterTemplate.objects.get(pk=template_id)
157 except ParameterTemplate.DoesNotExist:
158 raise ValueError(f'Invalid parameter template ID: {template_id}')
160 # Construct a "numeric" value for the filter
161 try:
162 value_numeric = float(value)
163 except (ValueError, TypeError):
164 value_numeric = None
166 if template.checkbox:
167 # Account for 'boolean' parameter values
168 bool_value = InvenTree.helpers.str2bool(value)
169 value_numeric = 1 if bool_value else 0
170 value = str(bool_value)
172 # Boolean filtering is limited to exact matches
173 func = ''
175 elif value_numeric is None and template.units:
176 # Convert the raw value to the units of the template parameter
177 try:
178 value_numeric = InvenTree.conversion.convert_physical_value(
179 value, template.units
180 )
181 except Exception:
182 # The value cannot be converted - return an empty queryset
183 return queryset.none()
185 # Special handling for the "not equal" operator
186 if func == 'ne':
187 invert = True
188 func = ''
189 else:
190 invert = False
192 # Some filters are only applicable to string values
193 text_only = any([func in ['icontains'], value_numeric is None])
195 # Ensure the function starts with a double underscore
196 if func and not func.startswith('__'):
197 func = f'__{func}'
199 # Query for 'numeric' value - this has priority over 'string' value
200 data_numeric = {
201 'parameters_list__template': template,
202 'parameters_list__data_numeric__isnull': False,
203 f'parameters_list__data_numeric{func}': value_numeric,
204 }
206 query_numeric = Q(**data_numeric)
208 # Query for 'string' value
209 data_text = {
210 'parameters_list__template': template,
211 f'parameters_list__data{func}': str(value),
212 }
214 if not text_only:
215 data_text['parameters_list__data_numeric__isnull'] = True
217 query_text = Q(**data_text)
219 # Combine the queries based on whether we are filtering by text or numeric value
220 q = query_text if text_only else query_text | query_numeric
222 # Special handling for the '__ne' (not equal) operator
223 # In this case, we want the *opposite* of the above queries
224 if invert:
225 return queryset.exclude(q).distinct()
226 else:
227 return queryset.filter(q).distinct()
230def filter_parametric_data(queryset: QuerySet, parameters: dict[str, str]) -> QuerySet:
231 """Filter the provided queryset by parametric data.
233 Arguments:
234 queryset: The initial queryset to filter.
235 parameters: A dictionary of parameter filters to apply.
237 Returns:
238 Filtered queryset.
240 Used to filter returned parts based on their parameter values.
242 To filter based on parameter value, supply query parameters like:
243 - parameter_<x>=<value>
244 - parameter_<x>_gt=<value>
245 - parameter_<x>_lte=<value>
247 where:
248 - <x> is the ID of the ParameterTemplate.
249 - <value> is the value to filter against.
251 Typically these filters would be provided against via an API request.
252 """
253 # Allowed lookup operations for parameter values
254 operators = '|'.join(PARAMETER_FILTER_OPERATORS)
256 regex_pattern = rf'^parameter_(\d+)(_({operators}))?$'
258 for param, value in parameters.items():
259 result = re.match(regex_pattern, param)
260 if not result: 260 ↛ 263line 260 didn't jump to line 263 because the condition on line 260 was always true
261 continue
263 template_id = result.group(1)
264 operator = result.group(3) or ''
266 queryset = filter_parameters_by_value(
267 queryset, template_id, value, func=operator
268 )
270 return queryset
273def order_by_parameter(
274 queryset: QuerySet, model_type: Model, ordering: str | None
275) -> QuerySet:
276 """Order the provided queryset by a parameter value.
278 Arguments:
279 queryset: The initial queryset to order.
280 model_type: The model type of the items in the queryset.
281 ordering: The ordering string provided by the user.
283 Returns:
284 Ordered queryset.
286 Used to order returned parts based on their parameter values.
288 To order based on parameter value, supply an ordering string like:
289 - parameter_<x>
290 - -parameter_<x>
292 where:
293 - <x> is the ID of the ParameterTemplate.
294 - A leading '-' indicates descending order.
295 """
296 import common.models
298 if not ordering:
299 # No ordering provided - return the original queryset
300 return queryset
302 result = re.match(r'^-?parameter_(\d+)$', ordering)
304 if not result: 304 ↛ 308line 304 didn't jump to line 308 because the condition on line 304 was always true
305 # Ordering does not match the expected pattern - return the original queryset
306 return queryset
308 template_id = result.group(1)
309 ascending = not ordering.startswith('-')
311 template_exists_filter = common.models.Parameter.objects.filter(
312 template__id=template_id,
313 model_type=ContentType.objects.get_for_model(model_type),
314 model_id=OuterRef('id'),
315 )
317 queryset = queryset.annotate(parameter_exists=Exists(template_exists_filter))
319 # Annotate the queryset with the parameter value for the provided template
320 queryset = queryset.annotate(
321 parameter_value=Case(
322 When(
323 parameter_exists=True,
324 then=Subquery(
325 template_exists_filter.values('data')[:1], output_field=CharField()
326 ),
327 ),
328 default=Value('', output_field=CharField()),
329 ),
330 parameter_value_numeric=Case(
331 When(
332 parameter_exists=True,
333 then=Subquery(
334 template_exists_filter.values('data_numeric')[:1],
335 output_field=FloatField(),
336 ),
337 ),
338 default=Value(0, output_field=FloatField()),
339 ),
340 )
342 prefix = '' if ascending else '-'
344 return queryset.order_by(
345 '-parameter_exists',
346 f'{prefix}parameter_value_numeric',
347 f'{prefix}parameter_value',
348 )
351def enable_project_code_filter(default: bool = True):
352 """Add an optional 'project_code_detail' field to an API serializer.
354 Arguments:
355 filter_name: The name of the filter field.
356 default: If True, enable the filter by default.
358 If applied, this field will automatically prefetch the 'project_code' relationship.
359 """
360 from common.serializers import ProjectCodeSerializer
362 return InvenTree.serializers.OptionalField(
363 serializer_class=ProjectCodeSerializer,
364 serializer_kwargs={
365 'source': 'project_code',
366 'many': False,
367 'read_only': True,
368 'allow_null': True,
369 },
370 default_include=default,
371 filter_name='project_code_detail',
372 prefetch_fields=['project_code'],
373 )
376def enable_project_label_filter(default: bool = True):
377 """Add an optional 'project_code_label' field to an API serializer.
379 Arguments:
380 filter_name: The name of the filter field.
381 default: If True, enable the filter by default.
383 If applied, this field will automatically prefetch the 'project_code' relationship.
384 """
385 return InvenTree.serializers.OptionalField(
386 serializer_class=serializers.CharField,
387 serializer_kwargs={
388 'source': 'project_code.code',
389 'read_only': True,
390 'label': _('Project Code Label'),
391 'allow_null': True,
392 },
393 default_include=default,
394 filter_name='project_code_detail',
395 prefetch_fields=['project_code'],
396 )
399def enable_parameters_filter():
400 """Add an optional 'parameters' field to an API serializer.
402 Arguments:
403 source: The source field for the serializer.
404 filter_name: The name of the filter field.
405 default: If True, enable the filter by default.
407 If applied, this field will automatically annotate the queryset with parameter data.
408 """
409 from common.serializers import ParameterSerializer
411 return InvenTree.serializers.OptionalField(
412 serializer_class=ParameterSerializer,
413 serializer_kwargs={
414 'many': True,
415 'read_only': True,
416 'allow_null': True,
417 'required': False,
418 },
419 default_include=False,
420 filter_name='parameters',
421 prefetch_fields=[
422 'parameters_list',
423 'parameters_list__model_type',
424 'parameters_list__updated_by',
425 'parameters_list__template',
426 ],
427 )
430def enable_tags_filter(default: bool = False):
431 """Add an optional 'tags' field to an API serializer.
433 Arguments:
434 default: If True, enable the filter by default.
436 If applied, this field will automatically prefetch the 'tags' relationship.
437 """
438 return InvenTree.serializers.OptionalField(
439 serializer_class=TagListSerializerField,
440 serializer_kwargs={'required': False},
441 default_include=default,
442 filter_name='tags',
443 prefetch_fields=['tags', 'tagged_items', 'tagged_items__tag'],
444 )