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

1"""Custom API filters for InvenTree.""" 

2 

3import re 

4 

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 _ 

21 

22import django_filters.rest_framework.filters as rest_filters 

23from rest_framework import serializers 

24from taggit.serializers import TagListSerializerField 

25 

26import InvenTree.conversion 

27import InvenTree.helpers 

28import InvenTree.serializers 

29 

30 

31def determine_content_type(content_type: str | int | None) -> ContentType | None: 

32 """Determine a ContentType instance from a string or integer input. 

33 

34 Arguments: 

35 content_type: The content type to resolve (name or ID). 

36 

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 

42 

43 ct = None 

44 

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 

51 

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 

59 

60 else: 

61 # Next, try to resolve the content type via a model name 

62 ct = ContentType.objects.filter(model__iexact=content_type).first() 

63 

64 return ct 

65 

66 

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. 

71 

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. 

77 

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 

83 

84 ct = determine_content_type(content_type) 

85 

86 if ct is None: 

87 raise ValidationError(f'Invalid content type: {content_type}') 

88 

89 q = Q(**{f'{field_name}': ct}) 

90 

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}) 

93 

94 return queryset.filter(q) 

95 

96 

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. 

99 

100 Example usage in a FilterSet: 

101 tags = TagsFilter(label=_('Tags')) 

102 

103 Example query: 

104 ?tags=apple,banana → returns only items tagged with both 'apple' AND 'banana' 

105 """ 

106 

107 def __init__(self, *args, **kwargs): 

108 """Initialize the filter.""" 

109 if 'label' not in kwargs: 

110 kwargs['label'] = _('Tags') 

111 

112 super().__init__(*args, **kwargs) 

113 

114 def filter(self, qs, value): 

115 """Filter queryset to items matching all provided tag names.""" 

116 if not value: 

117 return qs 

118 

119 tag_names = [t.strip() for t in value.split(',') if t.strip()] 

120 

121 for tag in tag_names: 

122 qs = qs.filter(tags__name__iexact=tag) 

123 

124 return qs.distinct() 

125 

126 

127"""A list of valid operators for filtering part parameters.""" 

128PARAMETER_FILTER_OPERATORS: list[str] = ['gt', 'gte', 'lt', 'lte', 'ne', 'icontains'] 

129 

130 

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. 

135 

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). 

141 

142 Returns: 

143 A list of Parameter instances which match the given criteria. 

144 

145 Notes: 

146 - Parts which do not have a value for the given parameter are excluded. 

147 """ 

148 from common.models import ParameterTemplate 

149 

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}') 

153 

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}') 

159 

160 # Construct a "numeric" value for the filter 

161 try: 

162 value_numeric = float(value) 

163 except (ValueError, TypeError): 

164 value_numeric = None 

165 

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) 

171 

172 # Boolean filtering is limited to exact matches 

173 func = '' 

174 

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() 

184 

185 # Special handling for the "not equal" operator 

186 if func == 'ne': 

187 invert = True 

188 func = '' 

189 else: 

190 invert = False 

191 

192 # Some filters are only applicable to string values 

193 text_only = any([func in ['icontains'], value_numeric is None]) 

194 

195 # Ensure the function starts with a double underscore 

196 if func and not func.startswith('__'): 

197 func = f'__{func}' 

198 

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 } 

205 

206 query_numeric = Q(**data_numeric) 

207 

208 # Query for 'string' value 

209 data_text = { 

210 'parameters_list__template': template, 

211 f'parameters_list__data{func}': str(value), 

212 } 

213 

214 if not text_only: 

215 data_text['parameters_list__data_numeric__isnull'] = True 

216 

217 query_text = Q(**data_text) 

218 

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 

221 

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() 

228 

229 

230def filter_parametric_data(queryset: QuerySet, parameters: dict[str, str]) -> QuerySet: 

231 """Filter the provided queryset by parametric data. 

232 

233 Arguments: 

234 queryset: The initial queryset to filter. 

235 parameters: A dictionary of parameter filters to apply. 

236 

237 Returns: 

238 Filtered queryset. 

239 

240 Used to filter returned parts based on their parameter values. 

241 

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> 

246 

247 where: 

248 - <x> is the ID of the ParameterTemplate. 

249 - <value> is the value to filter against. 

250 

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) 

255 

256 regex_pattern = rf'^parameter_(\d+)(_({operators}))?$' 

257 

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 

262 

263 template_id = result.group(1) 

264 operator = result.group(3) or '' 

265 

266 queryset = filter_parameters_by_value( 

267 queryset, template_id, value, func=operator 

268 ) 

269 

270 return queryset 

271 

272 

273def order_by_parameter( 

274 queryset: QuerySet, model_type: Model, ordering: str | None 

275) -> QuerySet: 

276 """Order the provided queryset by a parameter value. 

277 

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. 

282 

283 Returns: 

284 Ordered queryset. 

285 

286 Used to order returned parts based on their parameter values. 

287 

288 To order based on parameter value, supply an ordering string like: 

289 - parameter_<x> 

290 - -parameter_<x> 

291 

292 where: 

293 - <x> is the ID of the ParameterTemplate. 

294 - A leading '-' indicates descending order. 

295 """ 

296 import common.models 

297 

298 if not ordering: 

299 # No ordering provided - return the original queryset 

300 return queryset 

301 

302 result = re.match(r'^-?parameter_(\d+)$', ordering) 

303 

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 

307 

308 template_id = result.group(1) 

309 ascending = not ordering.startswith('-') 

310 

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 ) 

316 

317 queryset = queryset.annotate(parameter_exists=Exists(template_exists_filter)) 

318 

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 ) 

341 

342 prefix = '' if ascending else '-' 

343 

344 return queryset.order_by( 

345 '-parameter_exists', 

346 f'{prefix}parameter_value_numeric', 

347 f'{prefix}parameter_value', 

348 ) 

349 

350 

351def enable_project_code_filter(default: bool = True): 

352 """Add an optional 'project_code_detail' field to an API serializer. 

353 

354 Arguments: 

355 filter_name: The name of the filter field. 

356 default: If True, enable the filter by default. 

357 

358 If applied, this field will automatically prefetch the 'project_code' relationship. 

359 """ 

360 from common.serializers import ProjectCodeSerializer 

361 

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 ) 

374 

375 

376def enable_project_label_filter(default: bool = True): 

377 """Add an optional 'project_code_label' field to an API serializer. 

378 

379 Arguments: 

380 filter_name: The name of the filter field. 

381 default: If True, enable the filter by default. 

382 

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 ) 

397 

398 

399def enable_parameters_filter(): 

400 """Add an optional 'parameters' field to an API serializer. 

401 

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. 

406 

407 If applied, this field will automatically annotate the queryset with parameter data. 

408 """ 

409 from common.serializers import ParameterSerializer 

410 

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 ) 

428 

429 

430def enable_tags_filter(default: bool = False): 

431 """Add an optional 'tags' field to an API serializer. 

432 

433 Arguments: 

434 default: If True, enable the filter by default. 

435 

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 )