inventree/InvenTree
2
0
Fork 0
mirror of https://github.com/inventree/InvenTree.git synced 2025-09-13 22:21:37 +00:00
Code Activity

Merge remote-tracking branch 'inventree/master' into locate-mixin

Browse Source 
# Conflicts:
#	InvenTree/InvenTree/api.py
#	InvenTree/InvenTree/urls.py
#	InvenTree/plugin/base/integration/mixins.py
#	InvenTree/plugin/mixins/__init__.py
This commit is contained in:
Oliver Walters
2022-05-16 00:10:38 +10:00
parent fa954b3812 a26840d77f
commit 95dbfa1958
63 changed files with 1356 additions and 1132 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
InvenTree/plugin/base/__init__.py Normal file
View File

0
InvenTree/plugin/base/action/__init__.py Normal file
View File

45
InvenTree/plugin/base/action/api.py Normal file
View File

@@ -0,0 +1,45 @@
"""APIs for action plugins"""
from django.utils.translation import gettext_lazy as _
from rest_framework import permissions
from rest_framework.response import Response
from rest_framework.views import APIView
from plugin import registry
class ActionPluginView(APIView):
"""
Endpoint for running custom action plugins.
"""
permission_classes = [
permissions.IsAuthenticated,
]
def post(self, request, *args, **kwargs):
action = request.data.get('action', None)
data = request.data.get('data', None)
if action is None:
return Response({
'error': _("No action specified")
})
action_plugins = registry.with_mixin('action')
for plugin in action_plugins:
if plugin.action_name() == action:
# TODO @matmair use easier syntax once InvenTree 0.7.0 is released
plugin.init(request.user, data=data)
plugin.perform_action()
return Response(plugin.get_response())
# If we got to here, no matching action was found
return Response({
'error': _("No matching action found"),
"action": action,
})

69
InvenTree/plugin/base/action/mixins.py Normal file
View File

@@ -0,0 +1,69 @@
"""
Plugin mixin classes for action plugin
"""
class ActionMixin:
"""
Mixin that enables custom actions
"""
ACTION_NAME = ""
class MixinMeta:
"""
meta options for this mixin
"""
MIXIN_NAME = 'Actions'
def __init__(self, user=None, data=None):
super().__init__()
self.add_mixin('action', True, __class__)
self.init(user, data)
def action_name(self):
"""
Action name for this plugin.
If the ACTION_NAME parameter is empty,
uses the NAME instead.
"""
if self.ACTION_NAME:
return self.ACTION_NAME
return self.name
def init(self, user, data=None):
"""
An action plugin takes a user reference, and an optional dataset (dict)
"""
self.user = user
self.data = data
def perform_action(self):
"""
Override this method to perform the action!
"""
def get_result(self):
"""
Result of the action?
"""
# Re-implement this for cutsom actions
return False
def get_info(self):
"""
Extra info? Can be a string / dict / etc
"""
return None
def get_response(self):
"""
Return a response. Default implementation is a simple response
which can be overridden.
"""
return {
"action": self.action_name(),
"result": self.get_result(),
"info": self.get_info(),
}

94
InvenTree/plugin/base/action/test_action.py Normal file
View File

@@ -0,0 +1,94 @@
""" Unit tests for action plugins """
from django.test import TestCase
from django.contrib.auth import get_user_model
from plugin import InvenTreePlugin
from plugin.mixins import ActionMixin
class ActionMixinTests(TestCase):
""" Tests for ActionMixin """
ACTION_RETURN = 'a action was performed'
def setUp(self):
class SimplePlugin(ActionMixin, InvenTreePlugin):
pass
self.plugin = SimplePlugin('user')
class TestActionPlugin(ActionMixin, InvenTreePlugin):
"""a action plugin"""
ACTION_NAME = 'abc123'
def perform_action(self):
return ActionMixinTests.ACTION_RETURN + 'action'
def get_result(self):
return ActionMixinTests.ACTION_RETURN + 'result'
def get_info(self):
return ActionMixinTests.ACTION_RETURN + 'info'
self.action_plugin = TestActionPlugin('user')
class NameActionPlugin(ActionMixin, InvenTreePlugin):
NAME = 'Aplugin'
self.action_name = NameActionPlugin('user')
def test_action_name(self):
"""check the name definition possibilities"""
self.assertEqual(self.plugin.action_name(), '')
self.assertEqual(self.action_plugin.action_name(), 'abc123')
self.assertEqual(self.action_name.action_name(), 'Aplugin')
def test_function(self):
"""check functions"""
# the class itself
self.assertIsNone(self.plugin.perform_action())
self.assertEqual(self.plugin.get_result(), False)
self.assertIsNone(self.plugin.get_info())
self.assertEqual(self.plugin.get_response(), {
"action": '',
"result": False,
"info": None,
})
# overriden functions
self.assertEqual(self.action_plugin.perform_action(), self.ACTION_RETURN + 'action')
self.assertEqual(self.action_plugin.get_result(), self.ACTION_RETURN + 'result')
self.assertEqual(self.action_plugin.get_info(), self.ACTION_RETURN + 'info')
self.assertEqual(self.action_plugin.get_response(), {
"action": 'abc123',
"result": self.ACTION_RETURN + 'result',
"info": self.ACTION_RETURN + 'info',
})
class APITests(TestCase):
""" Tests for action api """
def setUp(self):
# Create a user for auth
user = get_user_model()
self.test_user = user.objects.create_user('testuser', 'test@testing.com', 'password')
self.client.login(username='testuser', password='password')
def test_post_errors(self):
"""Check the possible errors with post"""
# Test empty request
response = self.client.post('/api/action/')
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.data,
{'error': 'No action specified'}
)
# Test non-exsisting action
response = self.client.post('/api/action/', data={'action': "nonexsisting"})
self.assertEqual(response.status_code, 200)
self.assertEqual(
response.data,
{'error': 'No matching action found', 'action': 'nonexsisting'}
)

0
InvenTree/plugin/base/barcodes/__init__.py Normal file
View File

245
InvenTree/plugin/base/barcodes/api.py Normal file
View 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.base.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 = [
# Link a barcode to a part
path('link/', BarcodeAssign.as_view(), name='api-barcode-link'),
# Catch-all performs barcode 'scan'
re_path(r'^.*$', BarcodeScan.as_view(), name='api-barcode-scan'),
]

146
InvenTree/plugin/base/barcodes/mixins.py Normal file
View File

@@ -0,0 +1,146 @@
"""
Plugin mixin classes for barcode plugin
"""
import string
import hashlib
from stock.models import StockItem
from stock.serializers import StockItemSerializer, LocationSerializer
from part.serializers import PartSerializer
def hash_barcode(barcode_data):
"""
Calculate an MD5 hash of barcode data.
HACK: Remove any 'non printable' characters from the hash,
as it seems browers will remove special control characters...
TODO: Work out a way around this!
"""
barcode_data = str(barcode_data).strip()
printable_chars = filter(lambda x: x in string.printable, barcode_data)
barcode_data = ''.join(list(printable_chars))
result_hash = hashlib.md5(str(barcode_data).encode())
return str(result_hash.hexdigest())
class BarcodeMixin:
"""
Mixin that enables barcode handeling
Custom barcode plugins should use and extend this mixin as necessary.
"""
ACTION_NAME = ""
class MixinMeta:
"""
meta options for this mixin
"""
MIXIN_NAME = 'Barcode'
def __init__(self):
super().__init__()
self.add_mixin('barcode', 'has_barcode', __class__)
@property
def has_barcode(self):
"""
Does this plugin have everything needed to process a barcode
"""
return True
def init(self, barcode_data):
"""
Initialize the BarcodePlugin instance
Args:
barcode_data - The raw barcode data
"""
self.data = barcode_data
def getStockItem(self):
"""
Attempt to retrieve a StockItem associated with this barcode.
Default implementation returns None
"""
return None # pragma: no cover
def getStockItemByHash(self):
"""
Attempt to retrieve a StockItem associated with this barcode,
based on the barcode hash.
"""
try:
item = StockItem.objects.get(uid=self.hash())
return item
except StockItem.DoesNotExist:
return None
def renderStockItem(self, item):
"""
Render a stock item to JSON response
"""
serializer = StockItemSerializer(item, part_detail=True, location_detail=True, supplier_part_detail=True)
return serializer.data
def getStockLocation(self):
"""
Attempt to retrieve a StockLocation associated with this barcode.
Default implementation returns None
"""
return None # pragma: no cover
def renderStockLocation(self, loc):
"""
Render a stock location to a JSON response
"""
serializer = LocationSerializer(loc)
return serializer.data
def getPart(self):
"""
Attempt to retrieve a Part associated with this barcode.
Default implementation returns None
"""
return None # pragma: no cover
def renderPart(self, part):
"""
Render a part to JSON response
"""
serializer = PartSerializer(part)
return serializer.data
def hash(self):
"""
Calculate a hash for the barcode data.
This is supposed to uniquely identify the barcode contents,
at least within the bardcode sub-type.
The default implementation simply returns an MD5 hash of the barcode data,
encoded to a string.
This may be sufficient for most applications, but can obviously be overridden
by a subclass.
"""
return hash_barcode(self.data)
def validate(self):
"""
Default implementation returns False
"""
return False # pragma: no cover

266
InvenTree/plugin/base/barcodes/test_barcode.py Normal file
View File

@@ -0,0 +1,266 @@
# -*- 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_invalid_part(self):
"""Test response for invalid part"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'part': 999999999,
}
},
format='json'
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data['part'], 'Part does not exist')
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_invalid_item(self):
"""Test response for invalid stock item"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'stockitem': 999999999,
}
},
format='json'
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data['stockitem'], 'Stock item does not exist')
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_invalid_location(self):
"""Test response for an invalid location"""
response = self.client.post(
self.scan_url,
{
'barcode': {
'stocklocation': 999999999,
}
},
format='json'
)
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data['stocklocation'], 'Stock location does not exist')
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)

0
InvenTree/plugin/base/event/__init__.py Normal file
View File

192
InvenTree/plugin/base/event/events.py Normal file
View File

@@ -0,0 +1,192 @@
"""
Functions for triggering and responding to server side events
"""
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import logging
from django.conf import settings
from django.db import transaction
from django.db.models.signals import post_save, post_delete
from django.dispatch.dispatcher import receiver
from InvenTree.ready import canAppAccessDatabase, isImportingData
from InvenTree.tasks import offload_task
from plugin.registry import registry
logger = logging.getLogger('inventree')
def trigger_event(event, *args, **kwargs):
"""
Trigger an event with optional arguments.
This event will be stored in the database,
and the worker will respond to it later on.
"""
if not settings.PLUGINS_ENABLED:
# Do nothing if plugins are not enabled
return
if not canAppAccessDatabase():
logger.debug(f"Ignoring triggered event '{event}' - database not ready")
return
logger.debug(f"Event triggered: '{event}'")
offload_task(
'plugin.events.register_event',
event,
*args,
**kwargs
)
def register_event(event, *args, **kwargs):
"""
Register the event with any interested plugins.
Note: This function is processed by the background worker,
as it performs multiple database access operations.
"""
from common.models import InvenTreeSetting
logger.debug(f"Registering triggered event: '{event}'")
# Determine if there are any plugins which are interested in responding
if settings.PLUGIN_TESTING or InvenTreeSetting.get_setting('ENABLE_PLUGINS_EVENTS'):
with transaction.atomic():
for slug, plugin in registry.plugins.items():
if plugin.mixin_enabled('events'):
config = plugin.plugin_config()
if config and config.active:
logger.debug(f"Registering callback for plugin '{slug}'")
# Offload a separate task for each plugin
offload_task(
'plugin.events.process_event',
slug,
event,
*args,
**kwargs
)
def process_event(plugin_slug, event, *args, **kwargs):
"""
Respond to a triggered event.
This function is run by the background worker process.
This function may queue multiple functions to be handled by the background worker.
"""
logger.info(f"Plugin '{plugin_slug}' is processing triggered event '{event}'")
plugin = registry.plugins.get(plugin_slug, None)
if plugin is None:
logger.error(f"Could not find matching plugin for '{plugin_slug}'")
return
plugin.process_event(event, *args, **kwargs)
def allow_table_event(table_name):
"""
Determine if an automatic event should be fired for a given table.
We *do not* want events to be fired for some tables!
"""
if isImportingData():
# Prevent table events during the data import process
return False
table_name = table_name.lower().strip()
# Ignore any tables which start with these prefixes
ignore_prefixes = [
'account_',
'auth_',
'authtoken_',
'django_',
'error_',
'exchange_',
'otp_',
'plugin_',
'socialaccount_',
'user_',
'users_',
]
if any([table_name.startswith(prefix) for prefix in ignore_prefixes]):
return False
ignore_tables = [
'common_notificationentry',
'common_webhookendpoint',
'common_webhookmessage',
]
if table_name in ignore_tables:
return False
return True
@receiver(post_save)
def after_save(sender, instance, created, **kwargs):
"""
Trigger an event whenever a database entry is saved
"""
table = sender.objects.model._meta.db_table
instance_id = getattr(instance, 'id', None)
if instance_id is None:
return
if not allow_table_event(table):
return
if created:
trigger_event(
f'{table}.created',
id=instance.id,
model=sender.__name__,
)
else:
trigger_event(
f'{table}.saved',
id=instance.id,
model=sender.__name__,
)
@receiver(post_delete)
def after_delete(sender, instance, **kwargs):
"""
Trigger an event whenever a database entry is deleted
"""
table = sender.objects.model._meta.db_table
if not allow_table_event(table):
return
trigger_event(
f'{table}.deleted',
model=sender.__name__,
)

29
InvenTree/plugin/base/event/mixins.py Normal file
View File

@@ -0,0 +1,29 @@
"""Plugin mixin class for events"""
from plugin.helpers import MixinNotImplementedError
class EventMixin:
"""
Mixin that provides support for responding to triggered events.
Implementing classes must provide a "process_event" function:
"""
def process_event(self, event, *args, **kwargs):
"""
Function to handle events
Must be overridden by plugin
"""
# Default implementation does not do anything
raise MixinNotImplementedError
class MixinMeta:
"""
Meta options for this mixin
"""
MIXIN_NAME = 'Events'
def __init__(self):
super().__init__()
self.add_mixin('events', True, __class__)

0
InvenTree/plugin/base/integration/__init__.py Normal file
View File

601
InvenTree/plugin/base/integration/mixins.py Normal file
View File

@@ -0,0 +1,601 @@
"""
Plugin mixin classes
"""
import logging
import json
import requests
from django.urls import include, re_path
from django.db.utils import OperationalError, ProgrammingError
import InvenTree.helpers
from plugin.helpers import MixinImplementationError, MixinNotImplementedError, render_template
from plugin.models import PluginConfig, PluginSetting
from plugin.urls import PLUGIN_BASE
logger = logging.getLogger('inventree')
class SettingsMixin:
"""
Mixin that enables global settings for the plugin
"""
class MixinMeta:
MIXIN_NAME = 'Settings'
def __init__(self):
super().__init__()
self.add_mixin('settings', 'has_settings', __class__)
self.settings = getattr(self, 'SETTINGS', {})
@property
def has_settings(self):
"""
Does this plugin use custom global settings
"""
return bool(self.settings)
def get_setting(self, key):
"""
Return the 'value' of the setting associated with this plugin
"""
return PluginSetting.get_setting(key, plugin=self)
def set_setting(self, key, value, user=None):
"""
Set plugin setting value by key
"""
try:
plugin, _ = PluginConfig.objects.get_or_create(key=self.plugin_slug(), name=self.plugin_name())
except (OperationalError, ProgrammingError): # pragma: no cover
plugin = None
if not plugin:
# Cannot find associated plugin model, return
return # pragma: no cover
PluginSetting.set_setting(key, value, user, plugin=plugin)
class ScheduleMixin:
"""
Mixin that provides support for scheduled tasks.
Implementing classes must provide a dict object called SCHEDULED_TASKS,
which provides information on the tasks to be scheduled.
SCHEDULED_TASKS = {
# Name of the task (will be prepended with the plugin name)
'test_server': {
'func': 'myplugin.tasks.test_server', # Python function to call (no arguments!)
'schedule': "I", # Schedule type (see django_q.Schedule)
'minutes': 30, # Number of minutes (only if schedule type = Minutes)
'repeats': 5, # Number of repeats (leave blank for 'forever')
},
'member_func': {
'func': 'my_class_func', # Note, without the 'dot' notation, it will call a class member function
'schedule': "H", # Once per hour
},
}
Note: 'schedule' parameter must be one of ['I', 'H', 'D', 'W', 'M', 'Q', 'Y']
Note: The 'func' argument can take two different forms:
- Dotted notation e.g. 'module.submodule.func' - calls a global function with the defined path
- Member notation e.g. 'my_func' (no dots!) - calls a member function of the calling class
"""
ALLOWABLE_SCHEDULE_TYPES = ['I', 'H', 'D', 'W', 'M', 'Q', 'Y']
# Override this in subclass model
SCHEDULED_TASKS = {}
class MixinMeta:
"""
Meta options for this mixin
"""
MIXIN_NAME = 'Schedule'
def __init__(self):
super().__init__()
self.scheduled_tasks = self.get_scheduled_tasks()
self.validate_scheduled_tasks()
self.add_mixin('schedule', 'has_scheduled_tasks', __class__)
def get_scheduled_tasks(self):
return getattr(self, 'SCHEDULED_TASKS', {})
@property
def has_scheduled_tasks(self):
"""
Are tasks defined for this plugin
"""
return bool(self.scheduled_tasks)
def validate_scheduled_tasks(self):
"""
Check that the provided scheduled tasks are valid
"""
if not self.has_scheduled_tasks:
raise MixinImplementationError("SCHEDULED_TASKS not defined")
for key, task in self.scheduled_tasks.items():
if 'func' not in task:
raise MixinImplementationError(f"Task '{key}' is missing 'func' parameter")
if 'schedule' not in task:
raise MixinImplementationError(f"Task '{key}' is missing 'schedule' parameter")
schedule = task['schedule'].upper().strip()
if schedule not in self.ALLOWABLE_SCHEDULE_TYPES:
raise MixinImplementationError(f"Task '{key}': Schedule '{schedule}' is not a valid option")
# If 'minutes' is selected, it must be provided!
if schedule == 'I' and 'minutes' not in task:
raise MixinImplementationError(f"Task '{key}' is missing 'minutes' parameter")
def get_task_name(self, key):
"""
Task name for key
"""
# Generate a 'unique' task name
slug = self.plugin_slug()
return f"plugin.{slug}.{key}"
def get_task_names(self):
"""
All defined task names
"""
# Returns a list of all task names associated with this plugin instance
return [self.get_task_name(key) for key in self.scheduled_tasks.keys()]
def register_tasks(self):
"""
Register the tasks with the database
"""
try:
from django_q.models import Schedule
for key, task in self.scheduled_tasks.items():
task_name = self.get_task_name(key)
if Schedule.objects.filter(name=task_name).exists():
# Scheduled task already exists - continue!
continue # pragma: no cover
logger.info(f"Adding scheduled task '{task_name}'")
func_name = task['func'].strip()
if '.' in func_name:
"""
Dotted notation indicates that we wish to run a globally defined function,
from a specified Python module.
"""
Schedule.objects.create(
name=task_name,
func=func_name,
schedule_type=task['schedule'],
minutes=task.get('minutes', None),
repeats=task.get('repeats', -1),
)
else:
"""
Non-dotted notation indicates that we wish to call a 'member function' of the calling plugin.
This is managed by the plugin registry itself.
"""
slug = self.plugin_slug()
Schedule.objects.create(
name=task_name,
func='plugin.registry.call_function',
args=f"'{slug}', '{func_name}'",
schedule_type=task['schedule'],
minutes=task.get('minutes', None),
repeats=task.get('repeats', -1),
)
except (ProgrammingError, OperationalError): # pragma: no cover
# Database might not yet be ready
logger.warning("register_tasks failed, database not ready")
def unregister_tasks(self):
"""
Deregister the tasks with the database
"""
try:
from django_q.models import Schedule
for key, task in self.scheduled_tasks.items():
task_name = self.get_task_name(key)
try:
scheduled_task = Schedule.objects.get(name=task_name)
scheduled_task.delete()
except Schedule.DoesNotExist:
pass
except (ProgrammingError, OperationalError): # pragma: no cover
# Database might not yet be ready
logger.warning("unregister_tasks failed, database not ready")
class UrlsMixin:
"""
Mixin that enables custom URLs for the plugin
"""
class MixinMeta:
"""
Meta options for this mixin
"""
MIXIN_NAME = 'URLs'
def __init__(self):
super().__init__()
self.add_mixin('urls', 'has_urls', __class__)
self.urls = self.setup_urls()
def setup_urls(self):
"""
Setup url endpoints for this plugin
"""
return getattr(self, 'URLS', None)
@property
def base_url(self):
"""
Base url for this plugin
"""
return f'{PLUGIN_BASE}/{self.slug}/'
@property
def internal_name(self):
"""
Internal url pattern name
"""
return f'plugin:{self.slug}:'
@property
def urlpatterns(self):
"""
Urlpatterns for this plugin
"""
if self.has_urls:
return re_path(f'^{self.slug}/', include((self.urls, self.slug)), name=self.slug)
return None
@property
def has_urls(self):
"""
Does this plugin use custom urls
"""
return bool(self.urls)
class NavigationMixin:
"""
Mixin that enables custom navigation links with the plugin
"""
NAVIGATION_TAB_NAME = None
NAVIGATION_TAB_ICON = "fas fa-question"
class MixinMeta:
"""
Meta options for this mixin
"""
MIXIN_NAME = 'Navigation Links'
def __init__(self):
super().__init__()
self.add_mixin('navigation', 'has_naviation', __class__)
self.navigation = self.setup_navigation()
def setup_navigation(self):
"""
Setup navigation links for this plugin
"""
nav_links = getattr(self, 'NAVIGATION', None)
if nav_links:
# check if needed values are configured
for link in nav_links:
if False in [a in link for a in ('link', 'name', )]:
raise MixinNotImplementedError('Wrong Link definition', link)
return nav_links
@property
def has_naviation(self):
"""
Does this plugin define navigation elements
"""
return bool(self.navigation)
@property
def navigation_name(self):
"""
Name for navigation tab
"""
name = getattr(self, 'NAVIGATION_TAB_NAME', None)
if not name:
name = self.human_name
return name
@property
def navigation_icon(self):
"""
Icon-name for navigation tab
"""
return getattr(self, 'NAVIGATION_TAB_ICON', "fas fa-question")
class AppMixin:
"""
Mixin that enables full django app functions for a plugin
"""
class MixinMeta:
"""m
Mta options for this mixin
"""
MIXIN_NAME = 'App registration'
def __init__(self):
super().__init__()
self.add_mixin('app', 'has_app', __class__)
@property
def has_app(self):
"""
This plugin is always an app with this plugin
"""
return True
class APICallMixin:
"""
Mixin that enables easier API calls for a plugin
Steps to set up:
1. Add this mixin before (left of) SettingsMixin and PluginBase
2. Add two settings for the required url and token/passowrd (use `SettingsMixin`)
3. Save the references to keys of the settings in `API_URL_SETTING` and `API_TOKEN_SETTING`
4. (Optional) Set `API_TOKEN` to the name required for the token by the external API - Defaults to `Bearer`
5. (Optional) Override the `api_url` property method if the setting needs to be extended
6. (Optional) Override `api_headers` to add extra headers (by default the token and Content-Type are contained)
7. Access the API in you plugin code via `api_call`
Example:
```
from plugin import InvenTreePlugin
from plugin.mixins import APICallMixin, SettingsMixin
class SampleApiCallerPlugin(APICallMixin, SettingsMixin, InvenTreePlugin):
'''
A small api call sample
'''
NAME = "Sample API Caller"
SETTINGS = {
'API_TOKEN': {
'name': 'API Token',
'protected': True,
},
'API_URL': {
'name': 'External URL',
'description': 'Where is your API located?',
'default': 'reqres.in',
},
}
API_URL_SETTING = 'API_URL'
API_TOKEN_SETTING = 'API_TOKEN'
def get_external_url(self):
'''
returns data from the sample endpoint
'''
return self.api_call('api/users/2')
```
"""
API_METHOD = 'https'
API_URL_SETTING = None
API_TOKEN_SETTING = None
API_TOKEN = 'Bearer'
class MixinMeta:
"""meta options for this mixin"""
MIXIN_NAME = 'API calls'
def __init__(self):
super().__init__()
self.add_mixin('api_call', 'has_api_call', __class__)
@property
def has_api_call(self):
"""Is the mixin ready to call external APIs?"""
if not bool(self.API_URL_SETTING):
raise MixinNotImplementedError("API_URL_SETTING must be defined")
if not bool(self.API_TOKEN_SETTING):
raise MixinNotImplementedError("API_TOKEN_SETTING must be defined")
return True
@property
def api_url(self):
return f'{self.API_METHOD}://{self.get_setting(self.API_URL_SETTING)}'
@property
def api_headers(self):
headers = {'Content-Type': 'application/json'}
if getattr(self, 'API_TOKEN_SETTING'):
headers[self.API_TOKEN] = self.get_setting(self.API_TOKEN_SETTING)
return headers
def api_build_url_args(self, arguments):
groups = []
for key, val in arguments.items():
groups.append(f'{key}={",".join([str(a) for a in val])}')
return f'?{"&".join(groups)}'
def api_call(self, endpoint, method: str = 'GET', url_args=None, data=None, headers=None, simple_response: bool = True, endpoint_is_url: bool = False):
if url_args:
endpoint += self.api_build_url_args(url_args)
if headers is None:
headers = self.api_headers
if endpoint_is_url:
url = endpoint
else:
url = f'{self.api_url}/{endpoint}'
# build kwargs for call
kwargs = {
'url': url,
'headers': headers,
}
if data:
kwargs['data'] = json.dumps(data)
# run command
response = requests.request(method, **kwargs)
# return
if simple_response:
return response.json()
return response
class PanelMixin:
"""
Mixin which allows integration of custom 'panels' into a particular page.
The mixin provides a number of key functionalities:
- Adds an (initially hidden) panel to the page
- Allows rendering of custom templated content to the panel
- Adds a menu item to the 'navbar' on the left side of the screen
- Allows custom javascript to be run when the panel is initially loaded
The PanelMixin class allows multiple panels to be returned for any page,
and also allows the plugin to return panels for many different pages.
Any class implementing this mixin must provide the 'get_custom_panels' method,
which dynamically returns the custom panels for a particular page.
This method is provided with:
- view : The View object which is being rendered
- request : The HTTPRequest object
Note that as this is called dynamically (per request),
then the actual panels returned can vary depending on the particular request or page
The 'get_custom_panels' method must return a list of dict objects, each with the following keys:
- title : The title of the panel, to appear in the sidebar menu
- description : Extra descriptive text (optional)
- icon : The icon to appear in the sidebar menu
- content : The HTML content to appear in the panel, OR
- content_template : A template file which will be rendered to produce the panel content
- javascript : The javascript content to be rendered when the panel is loade, OR
- javascript_template : A template file which will be rendered to produce javascript
e.g.
{
'title': "Updates",
'description': "Latest updates for this part",
'javascript': 'alert("You just loaded this panel!")',
'content': '<b>Hello world</b>',
}
"""
class MixinMeta:
MIXIN_NAME = 'Panel'
def __init__(self):
super().__init__()
self.add_mixin('panel', True, __class__)
def get_custom_panels(self, view, request):
""" This method *must* be implemented by the plugin class """
raise NotImplementedError(f"{__class__} is missing the 'get_custom_panels' method")
def get_panel_context(self, view, request, context):
"""
Build the context data to be used for template rendering.
Custom class can override this to provide any custom context data.
(See the example in "custom_panel_sample.py")
"""
# Provide some standard context items to the template for rendering
context['plugin'] = self
context['request'] = request
context['user'] = getattr(request, 'user', None)
context['view'] = view
try:
context['object'] = view.get_object()
except AttributeError:
pass
return context
def render_panels(self, view, request, context):
panels = []
# Construct an updated context object for template rendering
ctx = self.get_panel_context(view, request, context)
for panel in self.get_custom_panels(view, request):
content_template = panel.get('content_template', None)
javascript_template = panel.get('javascript_template', None)
if content_template:
# Render content template to HTML
panel['content'] = render_template(self, content_template, ctx)
if javascript_template:
# Render javascript template to HTML
panel['javascript'] = render_template(self, javascript_template, ctx)
# Check for required keys
required_keys = ['title', 'content']
if any([key not in panel for key in required_keys]):
logger.warning(f"Custom panel for plugin {__class__} is missing a required parameter")
continue
# Add some information on this plugin
panel['plugin'] = self
panel['slug'] = self.slug
# Add a 'key' for the panel, which is mostly guaranteed to be unique
panel['key'] = InvenTree.helpers.generateTestKey(self.slug + panel.get('title', 'panel'))
panels.append(panel)
return panels

246
InvenTree/plugin/base/integration/test_mixins.py Normal file
View File

@@ -0,0 +1,246 @@
""" Unit tests for base mixins for plugins """
from django.test import TestCase
from django.conf import settings
from django.urls import include, re_path
from django.contrib.auth import get_user_model
from plugin import InvenTreePlugin
from plugin.mixins import AppMixin, SettingsMixin, UrlsMixin, NavigationMixin, APICallMixin
from plugin.urls import PLUGIN_BASE
from plugin.helpers import MixinNotImplementedError
class BaseMixinDefinition:
def test_mixin_name(self):
# mixin name
self.assertIn(self.MIXIN_NAME, [item['key'] for item in self.mixin.registered_mixins])
# human name
self.assertIn(self.MIXIN_HUMAN_NAME, [item['human_name'] for item in self.mixin.registered_mixins])
class SettingsMixinTest(BaseMixinDefinition, TestCase):
MIXIN_HUMAN_NAME = 'Settings'
MIXIN_NAME = 'settings'
MIXIN_ENABLE_CHECK = 'has_settings'
TEST_SETTINGS = {'SETTING1': {'default': '123', }}
def setUp(self):
class SettingsCls(SettingsMixin, InvenTreePlugin):
SETTINGS = self.TEST_SETTINGS
self.mixin = SettingsCls()
class NoSettingsCls(SettingsMixin, InvenTreePlugin):
pass
self.mixin_nothing = NoSettingsCls()
user = get_user_model()
self.test_user = user.objects.create_user('testuser', 'test@testing.com', 'password')
self.test_user.is_staff = True
def test_function(self):
# settings variable
self.assertEqual(self.mixin.settings, self.TEST_SETTINGS)
# calling settings
# not existing
self.assertEqual(self.mixin.get_setting('ABCD'), '')
self.assertEqual(self.mixin_nothing.get_setting('ABCD'), '')
# right setting
self.mixin.set_setting('SETTING1', '12345', self.test_user)
self.assertEqual(self.mixin.get_setting('SETTING1'), '12345')
# no setting
self.assertEqual(self.mixin_nothing.get_setting(''), '')
class UrlsMixinTest(BaseMixinDefinition, TestCase):
MIXIN_HUMAN_NAME = 'URLs'
MIXIN_NAME = 'urls'
MIXIN_ENABLE_CHECK = 'has_urls'
def setUp(self):
class UrlsCls(UrlsMixin, InvenTreePlugin):
def test():
return 'ccc'
URLS = [re_path('testpath', test, name='test'), ]
self.mixin = UrlsCls()
class NoUrlsCls(UrlsMixin, InvenTreePlugin):
pass
self.mixin_nothing = NoUrlsCls()
def test_function(self):
plg_name = self.mixin.plugin_name()
# base_url
target_url = f'{PLUGIN_BASE}/{plg_name}/'
self.assertEqual(self.mixin.base_url, target_url)
# urlpattern
target_pattern = re_path(f'^{plg_name}/', include((self.mixin.urls, plg_name)), name=plg_name)
self.assertEqual(self.mixin.urlpatterns.reverse_dict, target_pattern.reverse_dict)
# resolve the view
self.assertEqual(self.mixin.urlpatterns.resolve('/testpath').func(), 'ccc')
self.assertEqual(self.mixin.urlpatterns.reverse('test'), 'testpath')
# no url
self.assertIsNone(self.mixin_nothing.urls)
self.assertIsNone(self.mixin_nothing.urlpatterns)
# internal name
self.assertEqual(self.mixin.internal_name, f'plugin:{self.mixin.slug}:')
class AppMixinTest(BaseMixinDefinition, TestCase):
MIXIN_HUMAN_NAME = 'App registration'
MIXIN_NAME = 'app'
MIXIN_ENABLE_CHECK = 'has_app'
def setUp(self):
class TestCls(AppMixin, InvenTreePlugin):
pass
self.mixin = TestCls()
def test_function(self):
# test that this plugin is in settings
self.assertIn('plugin.samples.integration', settings.INSTALLED_APPS)
class NavigationMixinTest(BaseMixinDefinition, TestCase):
MIXIN_HUMAN_NAME = 'Navigation Links'
MIXIN_NAME = 'navigation'
MIXIN_ENABLE_CHECK = 'has_naviation'
def setUp(self):
class NavigationCls(NavigationMixin, InvenTreePlugin):
NAVIGATION = [
{'name': 'aa', 'link': 'plugin:test:test_view'},
]
NAVIGATION_TAB_NAME = 'abcd1'
self.mixin = NavigationCls()
class NothingNavigationCls(NavigationMixin, InvenTreePlugin):
pass
self.nothing_mixin = NothingNavigationCls()
def test_function(self):
# check right configuration
self.assertEqual(self.mixin.navigation, [{'name': 'aa', 'link': 'plugin:test:test_view'}, ])
# navigation name
self.assertEqual(self.mixin.navigation_name, 'abcd1')
self.assertEqual(self.nothing_mixin.navigation_name, '')
def test_fail(self):
# check wrong links fails
with self.assertRaises(NotImplementedError):
class NavigationCls(NavigationMixin, InvenTreePlugin):
NAVIGATION = ['aa', 'aa']
NavigationCls()
class APICallMixinTest(BaseMixinDefinition, TestCase):
MIXIN_HUMAN_NAME = 'API calls'
MIXIN_NAME = 'api_call'
MIXIN_ENABLE_CHECK = 'has_api_call'
def setUp(self):
class MixinCls(APICallMixin, SettingsMixin, InvenTreePlugin):
NAME = "Sample API Caller"
SETTINGS = {
'API_TOKEN': {
'name': 'API Token',
'protected': True,
},
'API_URL': {
'name': 'External URL',
'description': 'Where is your API located?',
'default': 'reqres.in',
},
}
API_URL_SETTING = 'API_URL'
API_TOKEN_SETTING = 'API_TOKEN'
def get_external_url(self, simple: bool = True):
'''
returns data from the sample endpoint
'''
return self.api_call('api/users/2', simple_response=simple)
self.mixin = MixinCls()
class WrongCLS(APICallMixin, InvenTreePlugin):
pass
self.mixin_wrong = WrongCLS()
class WrongCLS2(APICallMixin, InvenTreePlugin):
API_URL_SETTING = 'test'
self.mixin_wrong2 = WrongCLS2()
def test_base_setup(self):
"""Test that the base settings work"""
# check init
self.assertTrue(self.mixin.has_api_call)
# api_url
self.assertEqual('https://reqres.in', self.mixin.api_url)
# api_headers
headers = self.mixin.api_headers
self.assertEqual(headers, {'Bearer': '', 'Content-Type': 'application/json'})
def test_args(self):
"""Test that building up args work"""
# api_build_url_args
# 1 arg
result = self.mixin.api_build_url_args({'a': 'b'})
self.assertEqual(result, '?a=b')
# more args
result = self.mixin.api_build_url_args({'a': 'b', 'c': 'd'})
self.assertEqual(result, '?a=b&c=d')
# list args
result = self.mixin.api_build_url_args({'a': 'b', 'c': ['d', 'e', 'f', ]})
self.assertEqual(result, '?a=b&c=d,e,f')
def test_api_call(self):
"""Test that api calls work"""
# api_call
result = self.mixin.get_external_url()
self.assertTrue(result)
self.assertIn('data', result,)
# api_call without json conversion
result = self.mixin.get_external_url(False)
self.assertTrue(result)
self.assertEqual(result.reason, 'OK')
# api_call with full url
result = self.mixin.api_call('https://reqres.in/api/users/2', endpoint_is_url=True)
self.assertTrue(result)
# api_call with post and data
result = self.mixin.api_call(
'api/users/',
data={"name": "morpheus", "job": "leader"},
method='POST'
)
self.assertTrue(result)
self.assertEqual(result['name'], 'morpheus')
# api_call with filter
result = self.mixin.api_call('api/users', url_args={'page': '2'})
self.assertTrue(result)
self.assertEqual(result['page'], 2)
def test_function_errors(self):
"""Test function errors"""
# wrongly defined plugins should not load
with self.assertRaises(MixinNotImplementedError):
self.mixin_wrong.has_api_call()
# cover wrong token setting
with self.assertRaises(MixinNotImplementedError):
self.mixin_wrong2.has_api_call()

0
InvenTree/plugin/base/label/__init__.py Normal file
View File

53
InvenTree/plugin/base/label/label.py Normal file
View File

@@ -0,0 +1,53 @@
"""Functions to print a label to a mixin printer"""
import logging
from django.utils.translation import gettext_lazy as _
from plugin.registry import registry
import common.notifications
logger = logging.getLogger('inventree')
def print_label(plugin_slug, label_image, label_instance=None, user=None):
"""
Print label with the provided plugin.
This task is nominally handled by the background worker.
If the printing fails (throws an exception) then the user is notified.
Arguments:
plugin_slug: The unique slug (key) of the plugin
label_image: A PIL.Image image object to be printed
"""
logger.info(f"Plugin '{plugin_slug}' is printing a label")
plugin = registry.plugins.get(plugin_slug, None)
if plugin is None:
logger.error(f"Could not find matching plugin for '{plugin_slug}'")
return
try:
plugin.print_label(label_image, width=label_instance.width, height=label_instance.height)
except Exception as e:
# Plugin threw an error - notify the user who attempted to print
ctx = {
'name': _('Label printing failed'),
'message': str(e),
}
logger.error(f"Label printing failed: Sending notification to user '{user}'")
# Throw an error against the plugin instance
common.notifications.trigger_notifaction(
plugin.plugin_config(),
'label.printing_failed',
targets=[user],
context=ctx,
delivery_methods=[common.notifications.UIMessageNotification]
)

39
InvenTree/plugin/base/label/mixins.py Normal file
View File

@@ -0,0 +1,39 @@
"""Plugin mixin classes for label plugins"""
from plugin.helpers import MixinNotImplementedError
class LabelPrintingMixin:
"""
Mixin which enables direct printing of stock labels.
Each plugin must provide a NAME attribute, which is used to uniquely identify the printer.
The plugin must also implement the print_label() function
"""
class MixinMeta:
"""
Meta options for this mixin
"""
MIXIN_NAME = 'Label printing'
def __init__(self): # pragma: no cover
super().__init__()
self.add_mixin('labels', True, __class__)
def print_label(self, label, **kwargs):
"""
Callback to print a single label
Arguments:
label: A black-and-white pillow Image object
kwargs:
length: The length of the label (in mm)
width: The width of the label (in mm)
"""
# Unimplemented (to be implemented by the particular plugin class)
MixinNotImplementedError('This Plugin must implement a `print_label` method')

82
InvenTree/plugin/base/locate/api.py Normal file
View File

@@ -0,0 +1,82 @@
"""API for location plugins"""
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from rest_framework import permissions
from rest_framework.exceptions import ParseError, NotFound
from rest_framework.response import Response
from rest_framework.views import APIView
from InvenTree.tasks import offload_task
from plugin import registry
from stock.models import StockItem, StockLocation
class LocatePluginView(APIView):
"""
Endpoint for using a custom plugin to identify or 'locate' a stock item or location
"""
permission_classes = [
permissions.IsAuthenticated,
]
def post(self, request, *args, **kwargs):
# Which plugin to we wish to use?
plugin = request.data.get('plugin', None)
if not plugin:
raise ParseError("'plugin' field must be supplied")
# Check that the plugin exists, and supports the 'locate' mixin
plugins = registry.with_mixin('locate')
if plugin not in [p.slug for p in plugins]:
raise ParseError(f"Plugin '{plugin}' is not installed, or does not support the location mixin")
# StockItem to identify
item_pk = request.data.get('item', None)
# StockLocation to identify
location_pk = request.data.get('location', None)
if not item_pk and not location_pk:
raise ParseError("Must supply either 'item' or 'location' parameter")
data = {
"success": "Identification plugin activated",
"plugin": plugin,
}
# StockItem takes priority
if item_pk:
try:
StockItem.objects.get(pk=item_pk)
offload_task('plugin.registry.call_function', plugin, 'locate_stock_item', item_pk)
data['item'] = item_pk
return Response(data)
except StockItem.DoesNotExist:
raise NotFound("StockItem matching PK '{item}' not found")
elif location_pk:
try:
StockLocation.objects.get(pk=location_pk)
offload_task('plugin.registry.call_function', plugin, 'locate_stock_location', location_pk)
data['location'] = location_pk
return Response(data)
except StockLocation.DoesNotExist:
raise NotFound("StockLocation matching PK {'location'} not found")
else:
raise NotFound()

75
InvenTree/plugin/base/locate/mixins.py Normal file
View File

@@ -0,0 +1,75 @@
"""Plugin mixin for locating stock items and locations"""
import logging
from plugin.helpers import MixinImplementationError
logger = logging.getLogger('inventree')
class LocateMixin:
"""
Mixin class which provides support for 'locating' inventory items,
for example identifying the location of a particular StockLocation.
Plugins could implement audible or visual cues to direct attention to the location,
with (for e.g.) LED strips or buzzers, or some other method.
The plugins may also be used to *deliver* a particular stock item to the user.
A class which implements this mixin may implement the following methods:
- locate_stock_item : Used to locate / identify a particular stock item
- locate_stock_location : Used to locate / identify a particular stock location
Refer to the default method implementations below for more information!
"""
class MixinMeta:
MIXIN_NAME = "Locate"
def __init__(self):
super().__init__()
self.add_mixin('locate', True, __class__)
def locate_stock_item(self, item_pk):
"""
Attempt to locate a particular StockItem
Arguments:
item_pk: The PK (primary key) of the StockItem to be located
The default implementation for locating a StockItem
attempts to locate the StockLocation where the item is located.
An attempt is only made if the StockItem is *in stock*
Note: A custom implemenation could always change this behaviour
"""
logger.info(f"LocateMixin: Attempting to locate StockItem pk={item_pk}")
from stock.models import StockItem
try:
item = StockItem.objects.get(pk=item_pk)
if item.in_stock and item.location is not None:
self.locate_stock_location(item.location.pk)
except StockItem.DoesNotExist:
logger.warning("LocateMixin: StockItem pk={item_pk} not found")
pass
def locate_stock_location(self, location_pk):
"""
Attempt to location a particular StockLocation
Arguments:
location_pk: The PK (primary key) of the StockLocation to be located
Note: The default implementation here does nothing!
"""
raise MixinImplementationError