Coverage for src/backend/InvenTree/part/filters.py: 86%

1"""Custom query filters for the Part app. 

2 

3The code here makes heavy use of subquery annotations! 

4 

5Useful References: 

6 

7- https://hansonkd.medium.com/the-dramatic-benefits-of-django-subqueries-and-annotations-4195e0dafb16 

8- https://pypi.org/project/django-sql-utils/ 

9- https://docs.djangoproject.com/en/4.0/ref/models/expressions/ 

10- https://stackoverflow.com/questions/42543978/django-1-11-annotating-a-subquery-aggregate 

11 

12""" 

13 

14from decimal import Decimal 

15from typing import Optional 

16 

17from django.db import models 

18from django.db.models import ( 

19 Case, 

20 DecimalField, 

21 ExpressionWrapper, 

22 F, 

23 FloatField, 

24 Func, 

25 IntegerField, 

26 OuterRef, 

27 Q, 

28 Subquery, 

29 Value, 

30 When, 

31) 

32from django.db.models.functions import Cast, Coalesce, Greatest 

33from django.db.models.query import QuerySet 

34 

35from sql_util.utils import SubquerySum 

36 

37import part.models 

38import stock.models 

39from build.status_codes import BuildStatusGroups 

40from order.status_codes import ( 

41 PurchaseOrderStatusGroups, 

42 SalesOrderStatusGroups, 

43 TransferOrderStatusGroups, 

44) 

45 

46 

47def annotate_in_production_quantity(reference: str = '') -> QuerySet: 

48 """Annotate the 'in production' quantity for each part in a queryset. 

49 

50 - Sum the 'quantity' field for all stock items which are 'in production' for each part. 

51 - This is the total quantity of "incomplete build outputs" for all active builds 

52 - This will return the same quantity as the 'quantity_in_production' method on the Part model 

53 

54 Arguments: 

55 reference: Reference to the part from the current queryset (default = '') 

56 """ 

57 building_filter = Q( 

58 is_building=True, build__status__in=BuildStatusGroups.ACTIVE_CODES 

59 ) 

60 

61 return Coalesce( 

62 SubquerySum(f'{reference}stock_items__quantity', filter=building_filter), 

63 Decimal(0), 

64 output_field=DecimalField(), 

65 ) 

66 

67 

68def annotate_scheduled_to_build_quantity(reference: str = '') -> QuerySet: 

69 """Annotate the 'scheduled to build' quantity for each part in a queryset. 

70 

71 - This is total scheduled quantity for all build orders which are 'active' 

72 - This may be different to the "in production" quantity 

73 - This will return the same quantity as the 'quantity_being_built' method no the Part model 

74 """ 

75 building_filter = Q(status__in=BuildStatusGroups.ACTIVE_CODES) 

76 

77 return Coalesce( 

78 SubquerySum( 

79 Greatest( 

80 ExpressionWrapper( 

81 Cast(F(f'{reference}builds__quantity'), output_field=IntegerField()) 

82 - Cast( 

83 F(f'{reference}builds__completed'), output_field=IntegerField() 

84 ), 

85 output_field=IntegerField(), 

86 ), 

87 0, 

88 ), 

89 filter=building_filter, 

90 ), 

91 0, 

92 output_field=IntegerField(), 

93 ) 

94 

95 

96def annotate_on_order_quantity(reference: str = '') -> QuerySet: 

97 """Annotate the 'on order' quantity for each part in a queryset. 

98 

99 Sum the 'remaining quantity' of each line item for any open purchase orders for each part: 

100 

101 - Purchase order must be 'active' or 'pending' 

102 - Received quantity must be less than line item quantity 

103 

104 Note that in addition to the 'quantity' on order, we must also take into account 'pack_quantity'. 

105 """ 

106 # Filter only 'active' purchase orders 

107 # Filter only line with outstanding quantity 

108 order_filter = Q( 

109 order__status__in=PurchaseOrderStatusGroups.OPEN, quantity__gt=F('received') 

110 ) 

111 

112 return Greatest( 

113 Coalesce( 

114 SubquerySum( 

115 ExpressionWrapper( 

116 F(f'{reference}supplier_parts__purchase_order_line_items__quantity') 

117 * F(f'{reference}supplier_parts__pack_quantity_native'), 

118 output_field=DecimalField(), 

119 ), 

120 filter=order_filter, 

121 ), 

122 Decimal(0), 

123 output_field=DecimalField(), 

124 ) 

125 - Coalesce( 

126 SubquerySum( 

127 ExpressionWrapper( 

128 F(f'{reference}supplier_parts__purchase_order_line_items__received') 

129 * F(f'{reference}supplier_parts__pack_quantity_native'), 

130 output_field=DecimalField(), 

131 ), 

132 filter=order_filter, 

133 ), 

134 Decimal(0), 

135 output_field=DecimalField(), 

136 ), 

137 Decimal(0), 

138 output_field=DecimalField(), 

139 ) 

140 

141 

142def annotate_total_stock(reference: str = '', filter: Optional[Q] = None) -> QuerySet: 

143 """Annotate 'total stock' quantity against a queryset. 

144 

145 - This function calculates the 'total stock' for a given part 

146 - Finds all stock items associated with each part (using the provided filter) 

147 - Aggregates the 'quantity' of each relevant stock item 

148 

149 Args: 

150 reference (str): The relationship reference of the part from the current model e.g. 'part' 

151 filter (Q): Q object which defines how to filter the stock items 

152 """ 

153 # Stock filter only returns 'in stock' items 

154 stock_filter = stock.models.StockItem.IN_STOCK_FILTER 

155 

156 if filter is not None: 

157 stock_filter &= filter 

158 

159 return Coalesce( 

160 SubquerySum(f'{reference}stock_items__quantity', filter=stock_filter), 

161 Decimal(0), 

162 output_field=models.DecimalField(), 

163 ) 

164 

165 

166def annotate_build_order_requirements(reference: str = '') -> QuerySet: 

167 """Annotate the total quantity of each part required for build orders. 

168 

169 - Only interested in 'active' build orders 

170 - We are looking for any BuildLine items which required this part (bom_item.sub_part) 

171 - We are interested in the 'quantity' of each BuildLine item 

172 

173 """ 

174 # Active build orders only 

175 build_filter = Q(build__status__in=BuildStatusGroups.ACTIVE_CODES) 

176 

177 return Coalesce( 

178 SubquerySum( 

179 ExpressionWrapper( 

180 F(f'{reference}used_in__build_lines__quantity') 

181 - F(f'{reference}used_in__build_lines__consumed'), 

182 output_field=DecimalField(), 

183 ), 

184 filter=build_filter, 

185 ), 

186 Decimal(0), 

187 output_field=models.DecimalField(), 

188 ) 

189 

190 

191def annotate_build_order_allocations(reference: str = '', location=None) -> QuerySet: 

192 """Annotate the total quantity of each part allocated to build orders. 

193 

194 - This function calculates the total part quantity allocated to open build orders 

195 - Finds all build order allocations for each part (using the provided filter) 

196 - Aggregates the 'allocated quantity' for each relevant build order allocation item 

197 

198 Arguments: 

199 reference: The relationship reference of the part from the current model 

200 location: If provided, only allocated stock items from this location are considered 

201 """ 

202 # Build filter only returns 'active' build orders 

203 build_filter = Q(build_line__build__status__in=BuildStatusGroups.ACTIVE_CODES) 

204 

205 if location is not None: 205 ↛ 208line 205 didn't jump to line 208 because the condition on line 205 was never true

206 # Filter by location (including any child locations) 

207 

208 build_filter &= Q( 

209 stock_item__location__tree_id=location.tree_id, 

210 stock_item__location__lft__gte=location.lft, 

211 stock_item__location__rght__lte=location.rght, 

212 stock_item__location__level__gte=location.level, 

213 ) 

214 

215 return Coalesce( 

216 SubquerySum( 

217 f'{reference}stock_items__allocations__quantity', filter=build_filter 

218 ), 

219 Decimal(0), 

220 output_field=models.DecimalField(), 

221 ) 

222 

223 

224def annotate_sales_order_requirements(reference: str = '') -> QuerySet: 

225 """Annotate the total quantity of each part required for sales orders. 

226 

227 - Only interested in 'active' sales orders 

228 - We are looking for any order lines which requires this part 

229 - We are interested in 'quantity'-'shipped' 

230 

231 """ 

232 # Order filter only returns incomplete shipments for open orders 

233 order_filter = Q(order__status__in=SalesOrderStatusGroups.OPEN) 

234 return Coalesce( 

235 SubquerySum(f'{reference}sales_order_line_items__quantity', filter=order_filter) 

236 - SubquerySum( 

237 f'{reference}sales_order_line_items__shipped', filter=order_filter 

238 ), 

239 Decimal(0), 

240 output_field=models.DecimalField(), 

241 ) 

242 

243 

244def annotate_sales_order_allocations(reference: str = '', location=None) -> QuerySet: 

245 """Annotate the total quantity of each part allocated to sales orders. 

246 

247 - This function calculates the total part quantity allocated to open sales orders" 

248 - Finds all sales order allocations for each part (using the provided filter) 

249 - Aggregates the 'allocated quantity' for each relevant sales order allocation item 

250 

251 Arguments: 

252 reference: The relationship reference of the part from the current model 

253 location: If provided, only allocated stock items from this location are considered 

254 """ 

255 # Order filter only returns incomplete shipments for open orders 

256 order_filter = Q( 

257 line__order__status__in=SalesOrderStatusGroups.OPEN, 

258 shipment__shipment_date=None, 

259 ) 

260 

261 if location is not None: 261 ↛ 264line 261 didn't jump to line 264 because the condition on line 261 was never true

262 # Filter by location (including any child locations) 

263 

264 order_filter &= Q( 

265 item__location__tree_id=location.tree_id, 

266 item__location__lft__gte=location.lft, 

267 item__location__rght__lte=location.rght, 

268 item__location__level__gte=location.level, 

269 ) 

270 

271 return Coalesce( 

272 SubquerySum( 

273 f'{reference}stock_items__sales_order_allocations__quantity', 

274 filter=order_filter, 

275 ), 

276 Decimal(0), 

277 output_field=models.DecimalField(), 

278 ) 

279 

280 

281def annotate_transfer_order_allocations(reference: str = '', location=None) -> QuerySet: 

282 """Annotate the total quantity of each part allocated to transfer orders. 

283 

284 - This function calculates the total part quantity allocated to open transfer orders" 

285 - Finds all transfer order allocations for each part (using the provided filter) 

286 - Aggregates the 'allocated quantity' for each relevant transfer order allocation item 

287 

288 Arguments: 

289 reference: The relationship reference of the part from the current model 

290 location: If provided, only allocated stock items from this location are considered 

291 """ 

292 # Order filter only returns open orders 

293 order_filter = Q(line__order__status__in=TransferOrderStatusGroups.OPEN) 

294 

295 if location is not None: 

296 # Filter by location (including any child locations) 

297 

298 order_filter &= Q( 

299 item__location__tree_id=location.tree_id, 

300 item__location__lft__gte=location.lft, 

301 item__location__rght__lte=location.rght, 

302 item__location__level__gte=location.level, 

303 ) 

304 

305 return Coalesce( 

306 SubquerySum( 

307 f'{reference}stock_items__transfer_order_allocations__quantity', 

308 filter=order_filter, 

309 ), 

310 Decimal(0), 

311 output_field=models.DecimalField(), 

312 ) 

313 

314 

315def variant_stock_query(reference: str = '', filter: Optional[Q] = None) -> QuerySet: 

316 """Create a queryset to retrieve all stock items for variant parts under the specified part. 

317 

318 - Useful for annotating a queryset with aggregated information about variant parts 

319 

320 Args: 

321 reference: The relationship reference of the part from the current model 

322 filter: Q object which defines how to filter the returned StockItem instances 

323 """ 

324 stock_filter = stock.models.StockItem.IN_STOCK_FILTER 

325 

326 if filter: 326 ↛ 327line 326 didn't jump to line 327 because the condition on line 326 was never true

327 stock_filter &= filter 

328 

329 return stock.models.StockItem.objects.filter( 

330 part__tree_id=OuterRef(f'{reference}tree_id'), 

331 part__lft__gt=OuterRef(f'{reference}lft'), 

332 part__rght__lt=OuterRef(f'{reference}rght'), 

333 ).filter(stock_filter) 

334 

335 

336def annotate_variant_quantity(subquery: Q, reference: str = 'quantity') -> QuerySet: 

337 """Create a subquery annotation for all variant part stock items on the given parent query. 

338 

339 Args: 

340 subquery: A 'variant_stock_query' Q object 

341 reference: The relationship reference of the variant stock items from the current queryset 

342 """ 

343 return Coalesce( 

344 Subquery( 

345 subquery 

346 .annotate( 

347 total=Func(F(reference), function='SUM', output_field=FloatField()) 

348 ) 

349 .values('total') 

350 .order_by() 

351 ), 

352 0, 

353 output_field=FloatField(), 

354 ) 

355 

356 

357def annotate_category_parts() -> QuerySet: 

358 """Construct a queryset annotation which returns the number of parts in a particular category. 

359 

360 - Includes parts in subcategories also 

361 - Requires subquery to perform annotation 

362 """ 

363 # Construct a subquery to provide all parts in this category and any subcategories: 

364 subquery = part.models.Part.objects.exclude(category=None).filter( 

365 category__tree_id=OuterRef('tree_id'), 

366 category__lft__gte=OuterRef('lft'), 

367 category__rght__lte=OuterRef('rght'), 

368 category__level__gte=OuterRef('level'), 

369 ) 

370 

371 return Coalesce( 

372 Subquery( 

373 subquery 

374 .annotate( 

375 total=Func(F('pk'), function='COUNT', output_field=IntegerField()) 

376 ) 

377 .values('total') 

378 .order_by() 

379 ), 

380 0, 

381 output_field=IntegerField(), 

382 ) 

383 

384 

385def annotate_default_location(reference: str = '') -> QuerySet: 

386 """Construct a queryset that finds the closest default location in the part's category tree. 

387 

388 If the part's category has its own default_location, this is returned. 

389 If not, the category tree is traversed until a value is found. 

390 """ 

391 subquery = part.models.PartCategory.objects.filter( 

392 tree_id=OuterRef(f'{reference}tree_id'), 

393 lft__lt=OuterRef(f'{reference}lft'), 

394 rght__gt=OuterRef(f'{reference}rght'), 

395 level__lte=OuterRef(f'{reference}level'), 

396 parent__isnull=False, 

397 default_location__isnull=False, 

398 ).order_by('-level') 

399 

400 return Coalesce( 

401 F(f'{reference}default_location'), 

402 Subquery(subquery.values('default_location')[:1]), 

403 Value(None), 

404 output_field=IntegerField(), 

405 ) 

406 

407 

408def annotate_sub_categories() -> QuerySet: 

409 """Construct a queryset annotation which returns the number of subcategories for each provided category.""" 

410 subquery = part.models.PartCategory.objects.filter( 

411 tree_id=OuterRef('tree_id'), 

412 lft__gt=OuterRef('lft'), 

413 rght__lt=OuterRef('rght'), 

414 level__gt=OuterRef('level'), 

415 ) 

416 

417 return Coalesce( 

418 Subquery( 

419 subquery 

420 .annotate( 

421 total=Func(F('pk'), function='COUNT', output_field=IntegerField()) 

422 ) 

423 .values('total') 

424 .order_by() 

425 ), 

426 0, 

427 output_field=IntegerField(), 

428 ) 

429 

430 

431def annotate_bom_item_can_build(queryset: QuerySet, reference: str = '') -> QuerySet: 

432 """Annotate the 'can_build' quantity for each BomItem in a queryset. 

433 

434 Arguments: 

435 queryset: A queryset of BomItem objects 

436 reference: Reference to the BomItem from the current queryset (default = '') 

437 

438 To do this we need to also annotate some other fields which are used in the calculation: 

439 

440 - total_in_stock: Total stock quantity for the part (may include variant stock) 

441 - available_stock: Total available stock quantity for the part 

442 - variant_stock: Total stock quantity for any variant parts 

443 - substitute_stock: Total stock quantity for any substitute parts 

444 

445 And then finally, annotate the 'can_build' quantity for each BomItem: 

446 """ 

447 # Pre-fetch the required related fields 

448 queryset = queryset.prefetch_related( 

449 f'{reference}sub_part', 

450 f'{reference}sub_part__stock_items', 

451 f'{reference}sub_part__stock_items__allocations', 

452 f'{reference}sub_part__stock_items__sales_order_allocations', 

453 f'{reference}substitutes', 

454 f'{reference}substitutes__part__stock_items', 

455 ) 

456 

457 # Queryset reference to the linked sub_part instance 

458 sub_part_ref = f'{reference}sub_part__' 

459 

460 # Apply some aliased annotations to the queryset 

461 queryset = queryset.annotate( 

462 # Total stock quantity (just for the sub_part itself) 

463 total_stock=annotate_total_stock(sub_part_ref), 

464 # Total allocated to sales orders 

465 allocated_to_sales_orders=annotate_sales_order_allocations(sub_part_ref), 

466 # Total allocated to build orders 

467 allocated_to_build_orders=annotate_build_order_allocations(sub_part_ref), 

468 ) 

469 

470 # Annotate the "available" stock, based on the total stock and allocations 

471 queryset = queryset.annotate( 

472 available_stock=Greatest( 

473 ExpressionWrapper( 

474 F('total_stock') 

475 - F('allocated_to_sales_orders') 

476 - F('allocated_to_build_orders'), 

477 output_field=models.DecimalField(), 

478 ), 

479 Decimal(0), 

480 output_field=models.DecimalField(), 

481 ) 

482 ) 

483 

484 # Annotate the total stock for any variant parts 

485 vq = variant_stock_query(reference=sub_part_ref) 

486 

487 queryset = queryset.alias( 

488 variant_stock_total=annotate_variant_quantity(vq, reference='quantity'), 

489 variant_bo_allocations=annotate_variant_quantity( 

490 vq, reference='sales_order_allocations__quantity' 

491 ), 

492 variant_so_allocations=annotate_variant_quantity( 

493 vq, reference='allocations__quantity' 

494 ), 

495 ) 

496 

497 # Annotate total variant stock 

498 queryset = queryset.annotate( 

499 available_variant_stock=Greatest( 

500 ExpressionWrapper( 

501 F('variant_stock_total') 

502 - F('variant_bo_allocations') 

503 - F('variant_so_allocations'), 

504 output_field=FloatField(), 

505 ), 

506 0, 

507 output_field=FloatField(), 

508 ) 

509 ) 

510 

511 # Account for substitute parts 

512 substitute_ref = f'{reference}substitutes__part__' 

513 

514 # Extract similar information for any 'substitute' parts 

515 queryset = queryset.alias( 

516 substitute_stock=annotate_total_stock(reference=substitute_ref), 

517 substitute_build_allocations=annotate_build_order_allocations( 

518 reference=substitute_ref 

519 ), 

520 substitute_sales_allocations=annotate_sales_order_allocations( 

521 reference=substitute_ref 

522 ), 

523 ) 

524 

525 # Calculate 'available_substitute_stock' field 

526 queryset = queryset.annotate( 

527 available_substitute_stock=Greatest( 

528 ExpressionWrapper( 

529 F('substitute_stock') 

530 - F('substitute_build_allocations') 

531 - F('substitute_sales_allocations'), 

532 output_field=models.DecimalField(), 

533 ), 

534 Decimal(0), 

535 output_field=models.DecimalField(), 

536 ) 

537 ) 

538 

539 # Now we can annotate the total "available" stock for the BomItem 

540 queryset = queryset.alias( 

541 total_stock=ExpressionWrapper( 

542 F('available_variant_stock') 

543 + F('available_substitute_stock') 

544 + F('available_stock'), 

545 output_field=FloatField(), 

546 ) 

547 ) 

548 

549 # And finally, we can annotate the 'can_build' quantity for each BomItem 

550 queryset = queryset.annotate( 

551 can_build=Greatest( 

552 ExpressionWrapper( 

553 Case( 

554 When(Q(quantity=0), then=Value(0)), 

555 default=(F('total_stock') - F('setup_quantity')) 

556 / (F('quantity') * (1.0 + F('attrition') / 100.0)), 

557 output_field=FloatField(), 

558 ), 

559 output_field=FloatField(), 

560 ), 

561 Decimal(0), 

562 output_field=FloatField(), 

563 ) 

564 ) 

565 

566 return queryset