Coverage for src/backend/InvenTree/plugin/base/barcodes/api.py: 35%

388 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 17:47 +0000

1"""API endpoints for barcode plugins.""" 

2 

3from django.db.models import F 

4from django.urls import include, path 

5from django.utils.translation import gettext_lazy as _ 

6 

7import structlog 

8from django_filters.rest_framework.filterset import FilterSet 

9from drf_spectacular.utils import extend_schema, extend_schema_view 

10from rest_framework import status 

11from rest_framework.exceptions import PermissionDenied, ValidationError 

12from rest_framework.generics import CreateAPIView 

13from rest_framework.response import Response 

14 

15import common.models 

16import InvenTree.permissions 

17import order.models 

18import plugin.base.barcodes.helper 

19import stock.models 

20from common.settings import get_global_setting 

21from InvenTree.api import BulkDeleteMixin 

22from InvenTree.exceptions import log_error 

23from InvenTree.filters import SEARCH_ORDER_FILTER 

24from InvenTree.helpers import hash_barcode 

25from InvenTree.mixins import ListAPI, RetrieveDestroyAPI 

26from plugin import PluginMixinEnum, registry 

27from users.permissions import check_user_permission 

28 

29from . import serializers as barcode_serializers 

30 

31logger = structlog.get_logger('inventree') 

32 

33 

34class BarcodeView(CreateAPIView): 

35 """Custom view class for handling a barcode scan.""" 

36 

37 # Default serializer class (can be overridden) 

38 serializer_class = barcode_serializers.BarcodeSerializer 

39 

40 def log_scan(self, request, response=None, result: bool = False): 

41 """Log a barcode scan to the database. 

42 

43 Arguments: 

44 request: HTTP request object 

45 response: Optional response data 

46 result: Boolean indicating success or failure of the scan 

47 """ 

48 from common.models import BarcodeScanResult 

49 

50 # Extract context data from the request 

51 context = {**request.GET.dict(), **request.POST.dict(), **request.data} 

52 

53 barcode = context.pop('barcode', '') 

54 

55 # Exit if storing barcode scans is disabled 

56 if not get_global_setting('BARCODE_STORE_RESULTS', backup=False, create=False): 56 ↛ 60line 56 didn't jump to line 60 because the condition on line 56 was always true

57 return 

58 

59 # Ensure that the response data is stringified first, otherwise cannot be JSON encoded 

60 if isinstance(response, dict): 

61 response = {key: str(value) for key, value in response.items()} 

62 elif response is None: 

63 pass 

64 else: 

65 response = str(response) 

66 

67 # Ensure that the context data is stringified first, otherwise cannot be JSON encoded 

68 if isinstance(context, dict): 

69 context = {key: str(value) for key, value in context.items()} 

70 elif context is None: 

71 pass 

72 else: 

73 context = str(context) 

74 

75 # Ensure data is not too long 

76 if len(barcode) > BarcodeScanResult.BARCODE_SCAN_MAX_LEN: 

77 barcode = barcode[: BarcodeScanResult.BARCODE_SCAN_MAX_LEN] 

78 

79 try: 

80 BarcodeScanResult.objects.create( 

81 data=barcode, 

82 user=request.user, 

83 endpoint=request.path, 

84 response=response, 

85 result=result, 

86 context=context, 

87 ) 

88 

89 # Ensure that we do not store too many scans 

90 max_scans = int(get_global_setting('BARCODE_RESULTS_MAX_NUM', create=False)) 

91 num_scans = BarcodeScanResult.objects.count() 

92 

93 if num_scans > max_scans: 

94 n = num_scans - max_scans 

95 old_scan_ids = list( 

96 BarcodeScanResult.objects 

97 .all() 

98 .order_by('timestamp') 

99 .values_list('pk', flat=True)[:n] 

100 ) 

101 BarcodeScanResult.objects.filter(pk__in=old_scan_ids).delete() 

102 except Exception: 

103 # Gracefully log error to database 

104 log_error(f'{self.__class__.__name__}.log_scan', scope='barcode') 

105 

106 def queryset(self): 

107 """This API view does not have a queryset.""" 

108 return None 

109 

110 # Default permission classes (can be overridden) 

111 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

112 

113 def create(self, request, *args, **kwargs): 

114 """Handle create method - override default create.""" 

115 serializer = self.get_serializer(data=request.data) 

116 

117 try: 

118 serializer.is_valid(raise_exception=True) 

119 except Exception as exc: 

120 self.log_scan(request, response={'error': str(exc)}, result=False) 

121 raise exc 

122 

123 data = serializer.validated_data 

124 

125 barcode = str(data.pop('barcode')).strip() 

126 

127 return self.handle_barcode(barcode, request, **data) 

128 

129 def handle_barcode(self, barcode: str, request, **kwargs): 

130 """Handle barcode scan. 

131 

132 Arguments: 

133 barcode: Raw barcode value 

134 request: HTTP request object 

135 

136 kwargs: 

137 Any custom fields passed by the specific serializer 

138 """ 

139 raise NotImplementedError( 

140 f'handle_barcode not implemented for {self.__class__}' 

141 ) 

142 

143 def scan_barcode(self, barcode: str, request, **kwargs): 

144 """Perform a generic 'scan' of the provided barcode data. 

145 

146 Check each loaded plugin, and return the first valid match 

147 """ 

148 plugins = registry.with_mixin(PluginMixinEnum.BARCODE) 

149 

150 # Look for a barcode plugin which knows how to deal with this barcode 

151 plugin = None 

152 response = {} 

153 

154 for current_plugin in plugins: 

155 try: 

156 result = current_plugin.scan(barcode, user=request.user, **kwargs) 

157 except PermissionDenied as exc: 

158 raise exc 

159 except Exception: 

160 log_error('BarcodeView.scan_barcode', plugin=current_plugin.slug) 

161 continue 

162 

163 if result is None: 163 ↛ 166line 163 didn't jump to line 166 because the condition on line 163 was always true

164 continue 

165 

166 if len(result) == 0: 

167 continue 

168 

169 if 'error' in result: 

170 logger.info( 

171 '%s.scan(...) returned an error: %s', 

172 current_plugin.__class__.__name__, 

173 result['error'], 

174 ) 

175 if not response: 

176 plugin = current_plugin 

177 response = result 

178 else: 

179 # Return the first successful match 

180 plugin = current_plugin 

181 response = result 

182 break 

183 

184 response['plugin'] = plugin.name if plugin else None 

185 response['barcode_data'] = barcode 

186 response['barcode_hash'] = hash_barcode(barcode) 

187 

188 return response 

189 

190 

191class BarcodeScan(BarcodeView): 

192 """Endpoint for handling generic barcode scan requests. 

193 

194 Barcode data are decoded by the client application, 

195 and sent to this endpoint (as a JSON object) for validation. 

196 

197 A barcode could follow the internal InvenTree barcode format, 

198 or it could match to a third-party barcode format (e.g. Digikey). 

199 """ 

200 

201 def handle_barcode(self, barcode: str, request, **kwargs): 

202 """Perform barcode scan action. 

203 

204 Arguments: 

205 barcode: Raw barcode value 

206 request: HTTP request object 

207 

208 kwargs: 

209 Any custom fields passed by the specific serializer 

210 """ 

211 response = self.scan_barcode(barcode, request, **kwargs) 

212 

213 if response['plugin'] is None: 213 ↛ 218line 213 didn't jump to line 218 because the condition on line 213 was always true

214 response['error'] = _('No match found for barcode data') 

215 self.log_scan(request, response, False) 

216 raise ValidationError(response) 

217 

218 response['success'] = _('Match found for barcode data') 

219 

220 # Log the scan result 

221 self.log_scan(request, response, True) 

222 

223 return Response(response) 

224 

225 

226@extend_schema_view( 

227 post=extend_schema(responses={200: barcode_serializers.BarcodeSerializer}) 

228) 

229class BarcodeGenerate(CreateAPIView): 

230 """Endpoint for generating a barcode for a database object. 

231 

232 The barcode is generated by the selected barcode plugin. 

233 """ 

234 

235 serializer_class = barcode_serializers.BarcodeGenerateSerializer 

236 

237 def queryset(self): 

238 """This API view does not have a queryset.""" 

239 return None 

240 

241 # Default permission classes (can be overridden) 

242 permission_classes = [InvenTree.permissions.IsAuthenticatedOrReadScope] 

243 

244 def create(self, request, *args, **kwargs): 

245 """Perform the barcode generation action.""" 

246 serializer = self.get_serializer(data=request.data) 

247 serializer.is_valid(raise_exception=True) 

248 

249 model = serializer.validated_data.get('model') 

250 pk = serializer.validated_data.get('pk') 

251 model_cls = plugin.base.barcodes.helper.get_supported_barcode_models_map().get( 

252 model, None 

253 ) 

254 

255 if model_cls is None: 

256 raise ValidationError({'error': _('Model is not supported')}) 

257 

258 try: 

259 model_instance = model_cls.objects.get(pk=pk) 

260 except model_cls.DoesNotExist: 

261 raise ValidationError({'error': _('Model instance not found')}) 

262 

263 barcode_data = plugin.base.barcodes.helper.generate_barcode(model_instance) 

264 

265 return Response({'barcode': barcode_data}, status=status.HTTP_200_OK) 

266 

267 

268class BarcodeAssign(BarcodeView): 

269 """Endpoint for assigning a barcode to a stock item. 

270 

271 - This only works if the barcode is not already associated with an object in the database 

272 - If the barcode does not match an object, then the barcode hash is assigned to the StockItem 

273 """ 

274 

275 serializer_class = barcode_serializers.BarcodeAssignSerializer 

276 

277 def handle_barcode(self, barcode: str, request, **kwargs): 

278 """Respond to a barcode assign request. 

279 

280 Checks inputs and assign barcode (hash) to StockItem. 

281 """ 

282 # Here we only check against 'InvenTree' plugins 

283 inventree_barcode_plugin = registry.get_plugin('inventreebarcode') 

284 

285 # First check if the provided barcode matches an existing database entry 

286 if inventree_barcode_plugin: 286 ↛ 298line 286 didn't jump to line 298 because the condition on line 286 was always true

287 result = inventree_barcode_plugin.scan(barcode, user=request.user, **kwargs) 

288 

289 if result is not None: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true

290 result['error'] = _('Barcode matches existing item') 

291 result['plugin'] = inventree_barcode_plugin.name 

292 result['barcode_data'] = barcode 

293 

294 result.pop('success', None) 

295 

296 raise ValidationError(result) 

297 

298 barcode_hash = hash_barcode(barcode) 

299 

300 valid_labels = [] 

301 

302 for model in plugin.base.barcodes.helper.get_supported_barcode_models(): 

303 label = model.barcode_model_type() 

304 valid_labels.append(label) 

305 

306 if instance := kwargs.get(label): 306 ↛ 308line 306 didn't jump to line 308 because the condition on line 306 was never true

307 # Check that the user has the required permission 

308 if not check_user_permission(request.user, model, 'change'): 

309 raise PermissionDenied({ 

310 'error': f'You do not have the required permissions for {model}' 

311 }) 

312 

313 instance.assign_barcode(barcode_data=barcode, barcode_hash=barcode_hash) 

314 

315 return Response({ 

316 'success': f'Assigned barcode to {label} instance', 

317 label: {'pk': instance.pk}, 

318 'barcode_data': barcode, 

319 'barcode_hash': barcode_hash, 

320 }) 

321 

322 # If we got here, it means that no valid model types were provided 

323 raise ValidationError({ 

324 'error': f"Missing data: provide one of '{valid_labels}'" 

325 }) 

326 

327 

328class BarcodeUnassign(BarcodeView): 

329 """Endpoint for unlinking / unassigning a custom barcode from a database object.""" 

330 

331 serializer_class = barcode_serializers.BarcodeUnassignSerializer 

332 

333 def create(self, request, *args, **kwargs): 

334 """Respond to a barcode unassign request.""" 

335 serializer = self.get_serializer(data=request.data) 

336 serializer.is_valid(raise_exception=True) 

337 data = serializer.validated_data 

338 

339 supported_models = plugin.base.barcodes.helper.get_supported_barcode_models() 

340 

341 supported_labels = [model.barcode_model_type() for model in supported_models] 

342 model_names = ', '.join(supported_labels) 

343 

344 matched_labels = [] 

345 

346 for label in supported_labels: 

347 if label in data: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true

348 matched_labels.append(label) 

349 

350 if len(matched_labels) == 0: 350 ↛ 355line 350 didn't jump to line 355 because the condition on line 350 was always true

351 raise ValidationError({ 

352 'error': f"Missing data: Provide one of '{model_names}'" 

353 }) 

354 

355 if len(matched_labels) > 1: 

356 raise ValidationError({ 

357 'error': f"Multiple conflicting fields: '{model_names}'" 

358 }) 

359 

360 # At this stage, we know that we have received a single valid field 

361 for model in supported_models: 

362 label = model.barcode_model_type() 

363 

364 if instance := data.get(label, None): 

365 # Check that the user has the required permission 

366 if not check_user_permission(request.user, model, 'change'): 

367 raise PermissionDenied({ 

368 'error': f'You do not have the required permissions for {model}' 

369 }) 

370 

371 # Unassign the barcode data from the model instance 

372 instance.unassign_barcode() 

373 

374 return Response({ 

375 'success': f'Barcode unassigned from {label} instance' 

376 }) 

377 

378 # If we get to this point, something has gone wrong! 

379 raise ValidationError({'error': 'Could not unassign barcode'}) 

380 

381 

382class BarcodePOAllocate(BarcodeView): 

383 """Endpoint for allocating parts to a purchase order by scanning their barcode. 

384 

385 Note that the scanned barcode may point to: 

386 

387 - A Part object 

388 - A ManufacturerPart object 

389 - A SupplierPart object 

390 """ 

391 

392 role_required = ['purchase_order.add'] 

393 

394 serializer_class = barcode_serializers.BarcodePOAllocateSerializer 

395 

396 def get_supplier_part( 

397 self, purchase_order, part=None, supplier_part=None, manufacturer_part=None 

398 ): 

399 """Return a single matching SupplierPart (or else raise an exception). 

400 

401 Arguments: 

402 purchase_order: PurchaseOrder object 

403 part: Part object (optional) 

404 supplier_part: SupplierPart object (optional) 

405 manufacturer_part: ManufacturerPart object (optional) 

406 

407 Returns: 

408 SupplierPart object 

409 

410 Raises: 

411 ValidationError if no matching SupplierPart is found 

412 

413 """ 

414 import company.models 

415 

416 supplier = purchase_order.supplier 

417 

418 supplier_parts = company.models.SupplierPart.objects.filter(supplier=supplier) 

419 

420 if not part and not supplier_part and not manufacturer_part: 

421 raise ValidationError(_('No matching part data found')) 

422 

423 if part and (part_id := part.get('pk', None)): 

424 supplier_parts = supplier_parts.filter(part__pk=part_id) 

425 

426 if supplier_part: 

427 if supplier_part_id := supplier_part.get('pk', None): 

428 supplier_parts = supplier_parts.filter(pk=supplier_part_id) 

429 

430 if manufacturer_part: 

431 if manufacturer_part_id := manufacturer_part.get('pk', None): 

432 supplier_parts = supplier_parts.filter( 

433 manufacturer_part__pk=manufacturer_part_id 

434 ) 

435 

436 if supplier_parts.count() == 0: 

437 raise ValidationError(_('No matching supplier parts found')) 

438 

439 if supplier_parts.count() > 1: 

440 raise ValidationError(_('Multiple matching supplier parts found')) 

441 

442 # At this stage, we have a single matching supplier part 

443 return supplier_parts.first() 

444 

445 def handle_barcode(self, barcode: str, request, **kwargs): 

446 """Scan the provided barcode data.""" 

447 # The purchase order is provided as part of the request 

448 purchase_order = kwargs.get('purchase_order') 

449 

450 response = self.scan_barcode(barcode, request, **kwargs) 

451 

452 if response['plugin'] is None: 

453 response['error'] = _('No matching plugin found for barcode data') 

454 

455 else: 

456 try: 

457 supplier_part = self.get_supplier_part( 

458 purchase_order, 

459 part=response.get('part', None), 

460 supplier_part=response.get('supplierpart', None), 

461 manufacturer_part=response.get('manufacturerpart', None), 

462 ) 

463 response['success'] = _('Matched supplier part') 

464 response['supplierpart'] = supplier_part.format_matched_response( 

465 user=request.user 

466 ) 

467 except ValidationError as e: 

468 response['error'] = str(e) 

469 

470 # TODO: Determine the 'quantity to order' for the supplier part 

471 

472 self.log_scan(request, response, 'success' in response) 

473 

474 if 'error' in response: 

475 raise ValidationError 

476 

477 return Response(response) 

478 

479 

480class BarcodePOReceive(BarcodeView): 

481 """Endpoint for handling receiving parts by scanning their barcode. 

482 

483 Barcode data are decoded by the client application, 

484 and sent to this endpoint (as a JSON object) for validation. 

485 

486 The barcode should follow a third-party barcode format (e.g. Digikey) 

487 and ideally contain order_number and quantity information. 

488 

489 The following parameters are available: 

490 

491 - barcode: The raw barcode data (required) 

492 - purchase_order: The purchase order containing the item to receive (optional) 

493 - location: The destination location for the received item (optional) 

494 """ 

495 

496 role_required = ['purchase_order.add'] 

497 

498 serializer_class = barcode_serializers.BarcodePOReceiveSerializer 

499 

500 def handle_barcode(self, barcode: str, request, **kwargs): 

501 """Handle a barcode scan for a purchase order item.""" 

502 logger.debug("BarcodePOReceive: scanned barcode - '%s'", barcode) 

503 

504 # Extract optional fields from the dataset 

505 supplier = kwargs.get('supplier') 

506 purchase_order = kwargs.get('purchase_order') 

507 location = kwargs.get('location') 

508 line_item = kwargs.get('line_item') 

509 auto_allocate = kwargs.get('auto_allocate', True) 

510 

511 # Extract location from PurchaseOrder, if available 

512 if not location and purchase_order: 512 ↛ 513line 512 didn't jump to line 513 because the condition on line 512 was never true

513 try: 

514 po = order.models.PurchaseOrder.objects.get(pk=purchase_order) 

515 if po.destination: 

516 location = po.destination.pk 

517 except Exception: 

518 pass 

519 

520 plugins = registry.with_mixin(PluginMixinEnum.BARCODE) 

521 

522 # Look for a barcode plugin which knows how to deal with this barcode 

523 plugin = None 

524 

525 response = {'barcode_data': barcode, 'barcode_hash': hash_barcode(barcode)} 

526 

527 internal_barcode_plugin = next( 

528 filter(lambda plugin: plugin.name == 'InvenTreeBarcode', plugins) 

529 ) 

530 

531 if result := internal_barcode_plugin.scan(barcode, user=request.user, **kwargs): 531 ↛ 532line 531 didn't jump to line 532 because the condition on line 531 was never true

532 if 'stockitem' in result: 

533 response['error'] = _('Item has already been received') 

534 self.log_scan(request, response, False) 

535 raise ValidationError(response) 

536 

537 # Now, look just for "supplier-barcode" plugins 

538 plugins = registry.with_mixin(PluginMixinEnum.SUPPLIER_BARCODE) 

539 

540 plugin_slug = None 

541 

542 plugin_response = None 

543 

544 plugin_error = None 

545 

546 no_supplier_plugin_error = [] 

547 

548 supplier_purchase_order = None 

549 

550 plugin_supplier = None 

551 

552 supplier_part = None 

553 

554 for current_plugin in plugins: 554 ↛ 555line 554 didn't jump to line 555 because the loop on line 554 never started

555 try: 

556 # Will either Output Debugresponse if No_Match is True or return the regular response if No_Match is False 

557 result = current_plugin.scan_receive_item( 

558 barcode, 

559 request.user, 

560 supplier=supplier, 

561 purchase_order=purchase_order, 

562 location=location, 

563 line_item=line_item, 

564 auto_allocate=auto_allocate, 

565 ) 

566 

567 except Exception: 

568 log_error('BarcodePOReceive.handle_barcode', plugin=current_plugin.slug) 

569 continue 

570 

571 no_match = result.get('no_match', True) 

572 

573 # No_Match Determines if it found a exact match for all the required fields from scan_recieve_item 

574 if no_match is True: 

575 supplier_found = False 

576 

577 try: 

578 plugin_slug = current_plugin.slug 

579 supplier_purchase_order = result.get('PO') 

580 plugin_supplier = result.get('supplier') 

581 supplier_part = result.get('supplier_part') 

582 except KeyError as e: 

583 log_error( 

584 f'BarcodePOReceive.handle_barcode debugresponse: KeyError {e}' 

585 ) 

586 continue 

587 

588 # Supplier does not have associated Supplier ID 

589 if plugin_supplier is None: 

590 no_supplier_plugin_error.append(plugin_slug) 

591 continue 

592 

593 # No Purchase Order or Supplier Part Found 

594 if supplier_purchase_order is None and supplier_part is None: 

595 continue 

596 

597 # Purchase Order exists and is found but Supplier part does not exist 

598 if supplier_purchase_order != None and supplier_part is None: 

599 # Supplier was Found 

600 supplier_found = True 

601 plugin_error = _('Purchase order Found\rNo supplier Part Match') 

602 

603 # Supplier Part is Found but Purchase Order does not exist 

604 elif supplier_purchase_order is None and supplier_part != None: 

605 # Supplier was Found 

606 supplier_found = True 

607 plugin_error = _('Supplier Part Found\rNo Purchase Order Match') 

608 

609 # Supplier for PO or Supplier part in barcode was found 

610 if supplier_found is True: 

611 # Adds info on for what was found in the barcode 

612 response['supplier_matches'] = { 

613 'purchase_order': supplier_purchase_order, 

614 'no_match': no_match, 

615 'supplier': plugin_supplier, 

616 'supplier_part': supplier_part, 

617 } 

618 

619 if 'error' in result: 

620 logger.info( 

621 '%s.scan_receive_item(...) returned an error: %s', 

622 current_plugin.__class__.__name__, 

623 result['error'], 

624 ) 

625 if not plugin_response: 

626 plugin = current_plugin 

627 plugin_response = result 

628 else: 

629 plugin = current_plugin 

630 plugin_response = result 

631 break 

632 

633 response['plugin'] = plugin.name if plugin else None 

634 

635 # If there is a plugin response, and there is a match (no_match = false), combine the dictionaries 

636 if plugin_response and plugin_response.get('no_match') is False: 636 ↛ 637line 636 didn't jump to line 637 because the condition on line 636 was never true

637 response = {**response, **plugin_response} 

638 elif no_supplier_plugin_error: 638 ↛ 639line 638 didn't jump to line 639 because the condition on line 638 was never true

639 response['no_supplier_plugin_error'] = no_supplier_plugin_error 

640 

641 # A plugin has not been found! 

642 if plugin is None: 642 ↛ 646line 642 didn't jump to line 646 because the condition on line 642 was always true

643 response['error'] = _('No plugin match for supplier barcode') 

644 

645 # A plugin was found, with a Error 

646 elif plugin_error: 

647 response['error'] = plugin_error 

648 

649 self.log_scan(request, response, 'success' in response) 

650 

651 if 'error' in response: 651 ↛ 654line 651 didn't jump to line 654 because the condition on line 651 was always true

652 raise ValidationError(response) 

653 

654 return Response(response) 

655 

656 

657class BarcodeSOAllocate(BarcodeView): 

658 """Endpoint for allocating stock to a sales order, by scanning barcode. 

659 

660 The scanned barcode should map to a StockItem object. 

661 

662 Additional fields can be passed to the endpoint: 

663 

664 - SalesOrder (Required) 

665 - Line Item 

666 - Shipment 

667 - Quantity 

668 """ 

669 

670 role_required = ['sales_order.add'] 

671 

672 serializer_class = barcode_serializers.BarcodeSOAllocateSerializer 

673 

674 def get_line_item(self, stock_item, **kwargs): 

675 """Return the matching line item for the provided stock item. 

676 

677 Raises: 

678 ValidationError: If no single matching line item is found 

679 """ 

680 # Extract sales order object (required field) 

681 sales_order = kwargs['sales_order'] 

682 

683 # Next, check if a line-item is provided (optional field) 

684 if line_item := kwargs.get('line'): 

685 return line_item 

686 

687 # If not provided, we need to find the correct line item 

688 parts = stock_item.part.get_ancestors(include_self=True) 

689 

690 # Find any matching line items for the stock item 

691 lines = order.models.SalesOrderLineItem.objects.filter( 

692 order=sales_order, part__in=parts, shipped__lte=F('quantity') 

693 ) 

694 

695 if lines.count() > 1: 

696 raise ValidationError(_('Multiple matching line items found')) 

697 

698 if lines.count() == 0: 

699 raise ValidationError(_('No matching line item found')) 

700 

701 return lines.first() 

702 

703 def get_shipment(self, **kwargs): 

704 """Extract the shipment from the provided kwargs, or guess. 

705 

706 Raises: 

707 ValidationError: If the shipment does not match the sales order 

708 """ 

709 sales_order = kwargs['sales_order'] 

710 

711 if shipment := kwargs.get('shipment'): 

712 if shipment.order != sales_order: 

713 raise ValidationError(_('Shipment does not match sales order')) 

714 

715 return shipment 

716 

717 shipments = order.models.SalesOrderShipment.objects.filter( 

718 order=sales_order, delivery_date=None 

719 ) 

720 

721 if shipments.count() == 1: 

722 return shipments.first() 

723 

724 # If shipment cannot be determined, return None 

725 return None 

726 

727 def handle_barcode(self, barcode: str, request, **kwargs): 

728 """Handle barcode scan for sales order allocation. 

729 

730 Arguments: 

731 barcode: Raw barcode data 

732 request: HTTP request object 

733 

734 kwargs: 

735 sales_order: SalesOrder ID value (required) 

736 line: SalesOrderLineItem ID value (optional) 

737 shipment: SalesOrderShipment ID value (optional) 

738 """ 

739 logger.debug("BarcodeSOAllocate: scanned barcode - '%s'", barcode) 

740 

741 response = self.scan_barcode(barcode, request, **kwargs) 

742 

743 if 'sales_order' not in kwargs: 

744 # SalesOrder ID *must* be provided 

745 response['error'] = _('No sales order provided') 

746 elif response['plugin'] is None: 

747 # Check that the barcode at least matches a plugin 

748 response['error'] = _('No matching plugin found for barcode data') 

749 else: 

750 try: 

751 stock_item_id = response['stockitem'].get('pk', None) 

752 stock_item = stock.models.StockItem.objects.get(pk=stock_item_id) 

753 except Exception: 

754 response['error'] = _('Barcode does not match an existing stock item') 

755 

756 if 'error' in response: 

757 self.log_scan(request, response, False) 

758 raise ValidationError(response) 

759 

760 # At this stage, we have a valid StockItem object 

761 

762 try: 

763 # Extract any other data from the kwargs 

764 # Note: This may raise a ValidationError at some point - we break on the first error 

765 sales_order = kwargs['sales_order'] 

766 line_item = self.get_line_item(stock_item, **kwargs) 

767 shipment = self.get_shipment(**kwargs) 

768 if stock_item is not None and line_item is not None: 

769 if stock_item.part != line_item.part: 

770 response['error'] = _('Stock item does not match line item') 

771 except ValidationError as e: 

772 response['error'] = str(e) 

773 

774 if 'error' in response: 

775 self.log_scan(request, response, False) 

776 raise ValidationError(response) 

777 

778 quantity = kwargs.get('quantity') 

779 

780 # Override quantity for serialized items 

781 if stock_item.serialized: 

782 quantity = 1 

783 

784 elif quantity is None: 

785 quantity = line_item.quantity - line_item.shipped 

786 quantity = min(quantity, stock_item.unallocated_quantity()) 

787 

788 response = { 

789 **response, 

790 'stock_item': stock_item.pk if stock_item else None, 

791 'part': stock_item.part.pk if stock_item else None, 

792 'sales_order': sales_order.pk if sales_order else None, 

793 'line_item': line_item.pk if line_item else None, 

794 'shipment': shipment.pk if shipment else None, 

795 'quantity': quantity, 

796 } 

797 

798 if stock_item is not None and quantity is not None: 

799 if stock_item.unallocated_quantity() < quantity: 

800 response['error'] = _('Insufficient stock available') 

801 

802 # If we have sufficient information, we can allocate the stock item 

803 elif all( 

804 x is not None for x in [line_item, sales_order, shipment, quantity] 

805 ): 

806 order.models.SalesOrderAllocation.objects.create( 

807 line=line_item, 

808 shipment=shipment, 

809 item=stock_item, 

810 quantity=quantity, 

811 ) 

812 

813 response['success'] = _('Stock item allocated to sales order') 

814 

815 else: 

816 response['error'] = _('Not enough information') 

817 response['action_required'] = True 

818 

819 self.log_scan(request, response, 'success' in response) 

820 

821 if 'error' in response: 

822 raise ValidationError(response) 

823 else: 

824 return Response(response) 

825 

826 

827class BarcodeScanResultMixin: 

828 """Mixin class for BarcodeScan API endpoints.""" 

829 

830 queryset = common.models.BarcodeScanResult.objects.all() 

831 serializer_class = barcode_serializers.BarcodeScanResultSerializer 

832 permission_classes = [InvenTree.permissions.IsStaffOrReadOnlyScope] 

833 

834 def get_queryset(self): 

835 """Return the queryset for the BarcodeScan API.""" 

836 queryset = super().get_queryset() 

837 

838 # Pre-fetch user data 

839 queryset = queryset.prefetch_related('user') 

840 

841 return queryset 

842 

843 

844class BarcodeScanResultFilter(FilterSet): 

845 """Custom filterset for the BarcodeScanResult API.""" 

846 

847 class Meta: 

848 """Meta class for the BarcodeScanResultFilter.""" 

849 

850 model = common.models.BarcodeScanResult 

851 fields = ['user', 'result'] 

852 

853 

854class BarcodeScanResultList(BarcodeScanResultMixin, BulkDeleteMixin, ListAPI): 

855 """List API endpoint for BarcodeScan objects.""" 

856 

857 filterset_class = BarcodeScanResultFilter 

858 filter_backends = SEARCH_ORDER_FILTER 

859 

860 ordering_fields = ['user', 'timestamp', 'endpoint', 'result'] 

861 

862 ordering = '-timestamp' 

863 

864 search_fields = ['data'] 

865 

866 

867class BarcodeScanResultDetail(BarcodeScanResultMixin, RetrieveDestroyAPI): 

868 """Detail endpoint for a BarcodeScan object.""" 

869 

870 

871barcode_api_urls = [ 

872 # Barcode scan history 

873 path( 

874 'history/', 

875 include([ 

876 path( 

877 '<int:pk>/', 

878 BarcodeScanResultDetail.as_view(), 

879 name='api-barcode-scan-result-detail', 

880 ), 

881 path( 

882 '', BarcodeScanResultList.as_view(), name='api-barcode-scan-result-list' 

883 ), 

884 ]), 

885 ), 

886 # Generate a barcode for a database object 

887 path('generate/', BarcodeGenerate.as_view(), name='api-barcode-generate'), 

888 # Link a third-party barcode to an item (e.g. Part / StockItem / etc) 

889 path('link/', BarcodeAssign.as_view(), name='api-barcode-link'), 

890 # Unlink a third-party barcode from an item 

891 path('unlink/', BarcodeUnassign.as_view(), name='api-barcode-unlink'), 

892 # Receive a purchase order item by scanning its barcode 

893 path('po-receive/', BarcodePOReceive.as_view(), name='api-barcode-po-receive'), 

894 # Allocate parts to a purchase order by scanning their barcode 

895 path('po-allocate/', BarcodePOAllocate.as_view(), name='api-barcode-po-allocate'), 

896 # Allocate stock to a sales order by scanning barcode 

897 path('so-allocate/', BarcodeSOAllocate.as_view(), name='api-barcode-so-allocate'), 

898 # Catch-all performs barcode 'scan' 

899 path('', BarcodeScan.as_view(), name='api-barcode-scan'), 

900]