mirror of
https://github.com/inventree/InvenTree.git
synced 2025-07-01 11:10:54 +00:00
move barcodes to plugin
This commit is contained in:
Matthias
245
InvenTree/plugin/barcode.py
Normal file
245
InvenTree/plugin/barcode.py
Normal file
@ -0,0 +1,245 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from django.urls import reverse, path, re_path
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from rest_framework.exceptions import ValidationError
|
||||
from rest_framework import permissions
|
||||
from rest_framework.response import Response
|
||||
from rest_framework.views import APIView
|
||||
|
||||
from stock.models import StockItem
|
||||
from stock.serializers import StockItemSerializer
|
||||
|
||||
from plugin.builtin.barcodes.inventree_barcode import InvenTreeBarcodePlugin
|
||||
from plugin.builtin.barcodes.mixins import hash_barcode
|
||||
from plugin import registry
|
||||
|
||||
|
||||
class BarcodeScan(APIView):
|
||||
"""
|
||||
Endpoint for handling generic barcode scan requests.
|
||||
|
||||
Barcode data are decoded by the client application,
|
||||
and sent to this endpoint (as a JSON object) for validation.
|
||||
|
||||
A barcode could follow the internal InvenTree barcode format,
|
||||
or it could match to a third-party barcode format (e.g. Digikey).
|
||||
|
||||
When a barcode is sent to the server, the following parameters must be provided:
|
||||
|
||||
- barcode: The raw barcode data
|
||||
|
||||
plugins:
|
||||
Third-party barcode formats may be supported using 'plugins'
|
||||
(more information to follow)
|
||||
|
||||
hashing:
|
||||
Barcode hashes are calculated using MD5
|
||||
|
||||
"""
|
||||
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated,
|
||||
]
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
"""
|
||||
Respond to a barcode POST request
|
||||
"""
|
||||
|
||||
data = request.data
|
||||
|
||||
if 'barcode' not in data:
|
||||
raise ValidationError({'barcode': _('Must provide barcode_data parameter')})
|
||||
|
||||
plugins = registry.with_mixin('barcode')
|
||||
|
||||
barcode_data = data.get('barcode')
|
||||
|
||||
# Ensure that the default barcode handler is installed
|
||||
plugins.append(InvenTreeBarcodePlugin())
|
||||
|
||||
# Look for a barcode plugin which knows how to deal with this barcode
|
||||
plugin = None
|
||||
|
||||
for current_plugin in plugins:
|
||||
# TODO @matmair make simpler after InvenTree 0.7.0 release
|
||||
current_plugin.init(barcode_data)
|
||||
|
||||
if current_plugin.validate():
|
||||
plugin = current_plugin
|
||||
break
|
||||
|
||||
match_found = False
|
||||
response = {}
|
||||
|
||||
response['barcode_data'] = barcode_data
|
||||
|
||||
# A plugin has been found!
|
||||
if plugin is not None:
|
||||
|
||||
# Try to associate with a stock item
|
||||
item = plugin.getStockItem()
|
||||
|
||||
if item is None:
|
||||
item = plugin.getStockItemByHash()
|
||||
|
||||
if item is not None:
|
||||
response['stockitem'] = plugin.renderStockItem(item)
|
||||
response['url'] = reverse('stock-item-detail', kwargs={'pk': item.id})
|
||||
match_found = True
|
||||
|
||||
# Try to associate with a stock location
|
||||
loc = plugin.getStockLocation()
|
||||
|
||||
if loc is not None:
|
||||
response['stocklocation'] = plugin.renderStockLocation(loc)
|
||||
response['url'] = reverse('stock-location-detail', kwargs={'pk': loc.id})
|
||||
match_found = True
|
||||
|
||||
# Try to associate with a part
|
||||
part = plugin.getPart()
|
||||
|
||||
if part is not None:
|
||||
response['part'] = plugin.renderPart(part)
|
||||
response['url'] = reverse('part-detail', kwargs={'pk': part.id})
|
||||
match_found = True
|
||||
|
||||
response['hash'] = plugin.hash()
|
||||
response['plugin'] = plugin.name
|
||||
|
||||
# No plugin is found!
|
||||
# However, the hash of the barcode may still be associated with a StockItem!
|
||||
else:
|
||||
result_hash = hash_barcode(barcode_data)
|
||||
|
||||
response['hash'] = result_hash
|
||||
response['plugin'] = None
|
||||
|
||||
# Try to look for a matching StockItem
|
||||
try:
|
||||
item = StockItem.objects.get(uid=result_hash)
|
||||
serializer = StockItemSerializer(item, part_detail=True, location_detail=True, supplier_part_detail=True)
|
||||
response['stockitem'] = serializer.data
|
||||
response['url'] = reverse('stock-item-detail', kwargs={'pk': item.id})
|
||||
match_found = True
|
||||
except StockItem.DoesNotExist:
|
||||
pass
|
||||
|
||||
if not match_found:
|
||||
response['error'] = _('No match found for barcode data')
|
||||
else:
|
||||
response['success'] = _('Match found for barcode data')
|
||||
|
||||
return Response(response)
|
||||
|
||||
|
||||
class BarcodeAssign(APIView):
|
||||
"""
|
||||
Endpoint for assigning a barcode to a stock item.
|
||||
|
||||
- This only works if the barcode is not already associated with an object in the database
|
||||
- If the barcode does not match an object, then the barcode hash is assigned to the StockItem
|
||||
"""
|
||||
|
||||
permission_classes = [
|
||||
permissions.IsAuthenticated
|
||||
]
|
||||
|
||||
def post(self, request, *args, **kwargs):
|
||||
|
||||
data = request.data
|
||||
|
||||
if 'barcode' not in data:
|
||||
raise ValidationError({'barcode': _('Must provide barcode_data parameter')})
|
||||
|
||||
if 'stockitem' not in data:
|
||||
raise ValidationError({'stockitem': _('Must provide stockitem parameter')})
|
||||
|
||||
barcode_data = data['barcode']
|
||||
|
||||
try:
|
||||
item = StockItem.objects.get(pk=data['stockitem'])
|
||||
except (ValueError, StockItem.DoesNotExist):
|
||||
raise ValidationError({'stockitem': _('No matching stock item found')})
|
||||
|
||||
plugins = registry.with_mixin('barcode')
|
||||
|
||||
plugin = None
|
||||
|
||||
for current_plugin in plugins:
|
||||
# TODO @matmair make simpler after InvenTree 0.7.0 release
|
||||
current_plugin.init(barcode_data)
|
||||
|
||||
if current_plugin.validate():
|
||||
plugin = current_plugin
|
||||
break
|
||||
|
||||
match_found = False
|
||||
|
||||
response = {}
|
||||
|
||||
response['barcode_data'] = barcode_data
|
||||
|
||||
# Matching plugin was found
|
||||
if plugin is not None:
|
||||
|
||||
result_hash = plugin.hash()
|
||||
response['hash'] = result_hash
|
||||
response['plugin'] = plugin.name
|
||||
|
||||
# Ensure that the barcode does not already match a database entry
|
||||
|
||||
if plugin.getStockItem() is not None:
|
||||
match_found = True
|
||||
response['error'] = _('Barcode already matches Stock Item')
|
||||
|
||||
if plugin.getStockLocation() is not None:
|
||||
match_found = True
|
||||
response['error'] = _('Barcode already matches Stock Location')
|
||||
|
||||
if plugin.getPart() is not None:
|
||||
match_found = True
|
||||
response['error'] = _('Barcode already matches Part')
|
||||
|
||||
if not match_found:
|
||||
item = plugin.getStockItemByHash()
|
||||
|
||||
if item is not None:
|
||||
response['error'] = _('Barcode hash already matches Stock Item')
|
||||
match_found = True
|
||||
|
||||
else:
|
||||
result_hash = hash_barcode(barcode_data)
|
||||
|
||||
response['hash'] = result_hash
|
||||
response['plugin'] = None
|
||||
|
||||
# Lookup stock item by hash
|
||||
try:
|
||||
item = StockItem.objects.get(uid=result_hash)
|
||||
response['error'] = _('Barcode hash already matches Stock Item')
|
||||
match_found = True
|
||||
except StockItem.DoesNotExist:
|
||||
pass
|
||||
|
||||
if not match_found:
|
||||
response['success'] = _('Barcode associated with Stock Item')
|
||||
|
||||
# Save the barcode hash
|
||||
item.uid = response['hash']
|
||||
item.save()
|
||||
|
||||
serializer = StockItemSerializer(item, part_detail=True, location_detail=True, supplier_part_detail=True)
|
||||
response['stockitem'] = serializer.data
|
||||
|
||||
return Response(response)
|
||||
|
||||
|
||||
barcode_api_urls = [
|
||||
|
||||
path('link/', BarcodeAssign.as_view(), name='api-barcode-link'),
|
||||
|
||||
# Catch-all performs barcode 'scan'
|
||||
re_path(r'^.*$', BarcodeScan.as_view(), name='api-barcode-scan'),
|
||||
]
|
||||
144
InvenTree/plugin/builtin/barcodes/inventree_barcode.py
Normal file
144
InvenTree/plugin/builtin/barcodes/inventree_barcode.py
Normal file
@ -0,0 +1,144 @@
|
||||
"""
|
||||
The InvenTreeBarcodePlugin validates barcodes generated by InvenTree itself.
|
||||
It can be used as a template for developing third-party barcode plugins.
|
||||
|
||||
The data format is very simple, and maps directly to database objects,
|
||||
via the "id" parameter.
|
||||
|
||||
Parsing an InvenTree barcode simply involves validating that the
|
||||
references model objects actually exist in the database.
|
||||
"""
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import json
|
||||
|
||||
import plugin.integration
|
||||
from .mixins import BarcodeMixin
|
||||
|
||||
from stock.models import StockItem, StockLocation
|
||||
from part.models import Part
|
||||
|
||||
from rest_framework.exceptions import ValidationError
|
||||
|
||||
|
||||
class InvenTreeBarcodePlugin(BarcodeMixin, plugin.integration.IntegrationPluginBase):
|
||||
|
||||
PLUGIN_NAME = "InvenTreeBarcode"
|
||||
|
||||
def validate(self):
|
||||
"""
|
||||
An "InvenTree" barcode must be a jsonnable-dict with the following tags:
|
||||
|
||||
{
|
||||
'tool': 'InvenTree',
|
||||
'version': <anything>
|
||||
}
|
||||
|
||||
"""
|
||||
|
||||
# The data must either be dict or be able to dictified
|
||||
if type(self.data) is dict:
|
||||
pass
|
||||
elif type(self.data) is str:
|
||||
try:
|
||||
self.data = json.loads(self.data)
|
||||
if type(self.data) is not dict:
|
||||
return False
|
||||
except json.JSONDecodeError:
|
||||
return False
|
||||
else:
|
||||
return False # pragma: no cover
|
||||
|
||||
# If any of the following keys are in the JSON data,
|
||||
# let's go ahead and assume that the code is a valid InvenTree one...
|
||||
|
||||
for key in ['tool', 'version', 'InvenTree', 'stockitem', 'stocklocation', 'part']:
|
||||
if key in self.data.keys():
|
||||
return True
|
||||
|
||||
return True
|
||||
|
||||
def getStockItem(self):
|
||||
|
||||
for k in self.data.keys():
|
||||
if k.lower() == 'stockitem':
|
||||
|
||||
data = self.data[k]
|
||||
|
||||
pk = None
|
||||
|
||||
# Initially try casting to an integer
|
||||
try:
|
||||
pk = int(data)
|
||||
except (TypeError, ValueError): # pragma: no cover
|
||||
pk = None
|
||||
|
||||
if pk is None: # pragma: no cover
|
||||
try:
|
||||
pk = self.data[k]['id']
|
||||
except (AttributeError, KeyError):
|
||||
raise ValidationError({k: "id parameter not supplied"})
|
||||
|
||||
try:
|
||||
item = StockItem.objects.get(pk=pk)
|
||||
return item
|
||||
except (ValueError, StockItem.DoesNotExist): # pragma: no cover
|
||||
raise ValidationError({k, "Stock item does not exist"})
|
||||
|
||||
return None
|
||||
|
||||
def getStockLocation(self):
|
||||
|
||||
for k in self.data.keys():
|
||||
if k.lower() == 'stocklocation':
|
||||
|
||||
pk = None
|
||||
|
||||
# First try simple integer lookup
|
||||
try:
|
||||
pk = int(self.data[k])
|
||||
except (TypeError, ValueError): # pragma: no cover
|
||||
pk = None
|
||||
|
||||
if pk is None: # pragma: no cover
|
||||
# Lookup by 'id' field
|
||||
try:
|
||||
pk = self.data[k]['id']
|
||||
except (AttributeError, KeyError):
|
||||
raise ValidationError({k: "id parameter not supplied"})
|
||||
|
||||
try:
|
||||
loc = StockLocation.objects.get(pk=pk)
|
||||
return loc
|
||||
except (ValueError, StockLocation.DoesNotExist): # pragma: no cover
|
||||
raise ValidationError({k, "Stock location does not exist"})
|
||||
|
||||
return None
|
||||
|
||||
def getPart(self):
|
||||
|
||||
for k in self.data.keys():
|
||||
if k.lower() == 'part':
|
||||
|
||||
pk = None
|
||||
|
||||
# Try integer lookup first
|
||||
try:
|
||||
pk = int(self.data[k])
|
||||
except (TypeError, ValueError): # pragma: no cover
|
||||
pk = None
|
||||
|
||||
if pk is None: # pragma: no cover
|
||||
try:
|
||||
pk = self.data[k]['id']
|
||||
except (AttributeError, KeyError):
|
||||
raise ValidationError({k, 'id parameter not supplied'})
|
||||
|
||||
try:
|
||||
part = Part.objects.get(pk=pk)
|
||||
return part
|
||||
except (ValueError, Part.DoesNotExist): # pragma: no cover
|
||||
raise ValidationError({k, 'Part does not exist'})
|
||||
|
||||
return None
|
||||
62
InvenTree/plugin/builtin/barcodes/test_inventree_barcode.py
Normal file
62
InvenTree/plugin/builtin/barcodes/test_inventree_barcode.py
Normal file
@ -0,0 +1,62 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Unit tests for InvenTreeBarcodePlugin"""
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.urls import reverse
|
||||
|
||||
from rest_framework.test import APITestCase
|
||||
from rest_framework import status
|
||||
|
||||
|
||||
class TestInvenTreeBarcode(APITestCase):
|
||||
|
||||
fixtures = [
|
||||
'category',
|
||||
'part',
|
||||
'location',
|
||||
'stock'
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
# Create a user for auth
|
||||
user = get_user_model()
|
||||
user.objects.create_user('testuser', 'test@testing.com', 'password')
|
||||
|
||||
self.client.login(username='testuser', password='password')
|
||||
|
||||
def test_errors(self):
|
||||
"""
|
||||
Test all possible error cases for assigment action
|
||||
"""
|
||||
|
||||
def test_assert_error(barcode_data):
|
||||
response = self.client.post(
|
||||
reverse('api-barcode-link'), format='json',
|
||||
data={
|
||||
'barcode': barcode_data,
|
||||
'stockitem': 521
|
||||
}
|
||||
)
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('error', response.data)
|
||||
|
||||
# test with already existing stock
|
||||
test_assert_error('{"stockitem": 521}')
|
||||
|
||||
# test with already existing stock location
|
||||
test_assert_error('{"stocklocation": 7}')
|
||||
|
||||
# test with already existing part location
|
||||
test_assert_error('{"part": 10004}')
|
||||
|
||||
# test with hash
|
||||
test_assert_error('{"blbla": 10004}')
|
||||
|
||||
def test_scan(self):
|
||||
"""
|
||||
Test that a barcode can be scanned
|
||||
"""
|
||||
|
||||
response = self.client.post(reverse('api-barcode-scan'), format='json', data={'barcode': 'blbla=10004'})
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('success', response.data)
|
||||
@ -7,7 +7,7 @@ from ..builtin.integration.mixins import APICallMixin, AppMixin, LabelPrintingMi
|
||||
from common.notifications import SingleNotificationMethod, BulkNotificationMethod
|
||||
|
||||
from ..builtin.action.mixins import ActionMixin
|
||||
from ..builtin.barcode.mixins import BarcodeMixin
|
||||
from ..builtin.barcodes.mixins import BarcodeMixin
|
||||
|
||||
__all__ = [
|
||||
'APICallMixin',
|
||||
|
||||
216
InvenTree/plugin/tests_barcode.py
Normal file
216
InvenTree/plugin/tests_barcode.py
Normal file
@ -0,0 +1,216 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
Unit tests for Barcode endpoints
|
||||
"""
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.urls import reverse
|
||||
|
||||
from rest_framework.test import APITestCase
|
||||
from rest_framework import status
|
||||
|
||||
from stock.models import StockItem
|
||||
|
||||
|
||||
class BarcodeAPITest(APITestCase):
|
||||
|
||||
fixtures = [
|
||||
'category',
|
||||
'part',
|
||||
'location',
|
||||
'stock'
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
# Create a user for auth
|
||||
user = get_user_model()
|
||||
user.objects.create_user('testuser', 'test@testing.com', 'password')
|
||||
|
||||
self.client.login(username='testuser', password='password')
|
||||
|
||||
self.scan_url = reverse('api-barcode-scan')
|
||||
self.assign_url = reverse('api-barcode-link')
|
||||
|
||||
def postBarcode(self, url, barcode):
|
||||
|
||||
return self.client.post(url, format='json', data={'barcode': str(barcode)})
|
||||
|
||||
def test_invalid(self):
|
||||
|
||||
# test scan url
|
||||
response = self.client.post(self.scan_url, format='json', data={})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
# test wrong assign urls
|
||||
response = self.client.post(self.assign_url, format='json', data={})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
response = self.client.post(self.assign_url, format='json', data={'barcode': '123'})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
response = self.client.post(self.assign_url, format='json', data={'barcode': '123', 'stockitem': '123'})
|
||||
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_empty(self):
|
||||
|
||||
response = self.postBarcode(self.scan_url, '')
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('error', data)
|
||||
|
||||
self.assertIn('barcode_data', data)
|
||||
self.assertIn('hash', data)
|
||||
self.assertIn('plugin', data)
|
||||
self.assertIsNone(data['plugin'])
|
||||
|
||||
def test_find_part(self):
|
||||
"""
|
||||
Test that we can lookup a part based on ID
|
||||
"""
|
||||
|
||||
response = self.client.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'part': 1,
|
||||
},
|
||||
},
|
||||
format='json',
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('part', response.data)
|
||||
self.assertIn('barcode_data', response.data)
|
||||
self.assertEqual(response.data['part']['pk'], 1)
|
||||
|
||||
def test_find_stock_item(self):
|
||||
"""
|
||||
Test that we can lookup a stock item based on ID
|
||||
"""
|
||||
|
||||
response = self.client.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'stockitem': 1,
|
||||
}
|
||||
},
|
||||
format='json',
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('stockitem', response.data)
|
||||
self.assertIn('barcode_data', response.data)
|
||||
self.assertEqual(response.data['stockitem']['pk'], 1)
|
||||
|
||||
def test_find_location(self):
|
||||
"""
|
||||
Test that we can lookup a stock location based on ID
|
||||
"""
|
||||
|
||||
response = self.client.post(
|
||||
self.scan_url,
|
||||
{
|
||||
'barcode': {
|
||||
'stocklocation': 1,
|
||||
},
|
||||
},
|
||||
format='json'
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
self.assertIn('stocklocation', response.data)
|
||||
self.assertIn('barcode_data', response.data)
|
||||
self.assertEqual(response.data['stocklocation']['pk'], 1)
|
||||
|
||||
def test_integer_barcode(self):
|
||||
|
||||
response = self.postBarcode(self.scan_url, '123456789')
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('error', data)
|
||||
|
||||
self.assertIn('barcode_data', data)
|
||||
self.assertIn('hash', data)
|
||||
self.assertIn('plugin', data)
|
||||
self.assertIsNone(data['plugin'])
|
||||
|
||||
def test_array_barcode(self):
|
||||
|
||||
response = self.postBarcode(self.scan_url, "['foo', 'bar']")
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
data = response.data
|
||||
self.assertIn('error', data)
|
||||
|
||||
self.assertIn('barcode_data', data)
|
||||
self.assertIn('hash', data)
|
||||
self.assertIn('plugin', data)
|
||||
self.assertIsNone(data['plugin'])
|
||||
|
||||
def test_barcode_generation(self):
|
||||
|
||||
item = StockItem.objects.get(pk=522)
|
||||
|
||||
response = self.postBarcode(self.scan_url, item.format_barcode())
|
||||
data = response.data
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
self.assertIn('stockitem', data)
|
||||
|
||||
pk = data['stockitem']['pk']
|
||||
|
||||
self.assertEqual(pk, item.pk)
|
||||
|
||||
def test_association(self):
|
||||
"""
|
||||
Test that a barcode can be associated with a StockItem
|
||||
"""
|
||||
|
||||
item = StockItem.objects.get(pk=522)
|
||||
|
||||
self.assertEqual(len(item.uid), 0)
|
||||
|
||||
barcode_data = 'A-TEST-BARCODE-STRING'
|
||||
|
||||
response = self.client.post(
|
||||
self.assign_url, format='json',
|
||||
data={
|
||||
'barcode': barcode_data,
|
||||
'stockitem': item.pk
|
||||
}
|
||||
)
|
||||
|
||||
data = response.data
|
||||
|
||||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||
|
||||
self.assertIn('success', data)
|
||||
|
||||
result_hash = data['hash']
|
||||
|
||||
# Read the item out from the database again
|
||||
item = StockItem.objects.get(pk=522)
|
||||
|
||||
self.assertEqual(result_hash, item.uid)
|
||||
|
||||
# Ensure that the same UID cannot be assigned to a different stock item!
|
||||
response = self.client.post(
|
||||
self.assign_url, format='json',
|
||||
data={
|
||||
'barcode': barcode_data,
|
||||
'stockitem': 521
|
||||
}
|
||||
)
|
||||
|
||||
data = response.data
|
||||
|
||||
self.assertIn('error', data)
|
||||
self.assertNotIn('success', data)
|
||||
Reference in New Issue
Block a user