Merge pull request #2518 from matmair/api-mixin

[Plugin] API Call mixin

InvenTree/plugin/__init__.py

@@ -1,3 +1,7 @@
+"""
+Utility file to enable simper imports
+"""
+
 from .registry import plugin_registry
 from .plugin import InvenTreePlugin
 from .integration import IntegrationPluginBase

InvenTree/plugin/builtin/integration/mixins.py

@@ -3,6 +3,8 @@ Plugin mixin classes
 """
 
 import logging
+import json
+import requests
 
 from django.conf.urls import url, include
 from django.db.utils import OperationalError, ProgrammingError
@@ -301,3 +303,113 @@ class AppMixin:
         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 IntegrationPluginBase
+    from plugin.mixins import APICallMixin, SettingsMixin
+
+
+    class SampleApiCallerPlugin(APICallMixin, SettingsMixin, IntegrationPluginBase):
+        '''
+        A small api call sample
+        '''
+        PLUGIN_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 ValueError("API_URL_SETTING must be defined")
+        if not bool(self.API_TOKEN_SETTING):
+            raise ValueError("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):
+        return {
+            self.API_TOKEN: self.get_setting(self.API_TOKEN_SETTING),
+            'Content-Type': 'application/json'
+        }
+
+    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):
+        if url_args:
+            endpoint += self.api_build_url_args(url_args)
+
+        if headers is None:
+            headers = self.api_headers
+
+        # build kwargs for call
+        kwargs = {
+            'url': f'{self.api_url}/{endpoint}',
+            '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

InvenTree/plugin/mixins/__init__.py

@@ -2,7 +2,7 @@
 Utility class to enable simpler imports
 """
 
-from ..builtin.integration.mixins import AppMixin, SettingsMixin, ScheduleMixin, UrlsMixin, NavigationMixin
+from ..builtin.integration.mixins import AppMixin, SettingsMixin, ScheduleMixin, UrlsMixin, NavigationMixin, APICallMixin
 
 __all__ = [
     'AppMixin',
@@ -10,4 +10,5 @@ __all__ = [
     'ScheduleMixin',
     'SettingsMixin',
     'UrlsMixin',
+    'APICallMixin',
 ]

InvenTree/plugin/samples/integration/api_caller.py

@@ -0,0 +1,32 @@
+"""
+Sample plugin for calling an external API
+"""
+from plugin import IntegrationPluginBase
+from plugin.mixins import APICallMixin, SettingsMixin
+
+
+class SampleApiCallerPlugin(APICallMixin, SettingsMixin, IntegrationPluginBase):
+    """
+    A small api call sample
+    """
+    PLUGIN_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')

InvenTree/plugin/samples/integration/test_api_caller.py

@@ -0,0 +1,21 @@
+""" Unit tests for action caller sample"""
+
+from django.test import TestCase
+
+from plugin import plugin_registry
+
+
+class SampleApiCallerPluginTests(TestCase):
+    """ Tests for SampleApiCallerPluginTests """
+
+    def test_return(self):
+        """check if the external api call works"""
+        # The plugin should be defined
+        self.assertIn('sample-api-caller', plugin_registry.plugins)
+        plg = plugin_registry.plugins['sample-api-caller']
+        self.assertTrue(plg)
+
+        # do an api call
+        result = plg.get_external_url()
+        self.assertTrue(result)
+        self.assertIn('data', result,)

InvenTree/plugin/test_integration.py

@@ -8,16 +8,16 @@ from django.contrib.auth import get_user_model
 from datetime import datetime
 
 from plugin import IntegrationPluginBase
-from plugin.mixins import AppMixin, SettingsMixin, UrlsMixin, NavigationMixin
+from plugin.mixins import AppMixin, SettingsMixin, UrlsMixin, NavigationMixin, APICallMixin
 from plugin.urls import PLUGIN_BASE
 
 
 class BaseMixinDefinition:
     def test_mixin_name(self):
         # mixin name
-        self.assertEqual(self.mixin.registered_mixins[0]['key'], self.MIXIN_NAME)
+        self.assertIn(self.MIXIN_NAME, [item['key'] for item in self.mixin.registered_mixins])
         # human name
-        self.assertEqual(self.mixin.registered_mixins[0]['human_name'], self.MIXIN_HUMAN_NAME)
+        self.assertIn(self.MIXIN_HUMAN_NAME, [item['human_name'] for item in self.mixin.registered_mixins])
 
 
 class SettingsMixinTest(BaseMixinDefinition, TestCase):
@@ -142,6 +142,79 @@ class NavigationMixinTest(BaseMixinDefinition, TestCase):
         self.assertEqual(self.nothing_mixin.navigation_name, '')
 
 
+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, IntegrationPluginBase):
+            PLUGIN_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')
+        self.mixin = MixinCls()
+
+        class WrongCLS(APICallMixin, IntegrationPluginBase):
+            pass
+        self.mixin_wrong = WrongCLS()
+
+        class WrongCLS2(APICallMixin, IntegrationPluginBase):
+            API_URL_SETTING = 'test'
+        self.mixin_wrong2 = WrongCLS2()
+
+    def test_function(self):
+        # 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'})
+
+        # 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')
+
+        # api_call
+        result = self.mixin.get_external_url()
+        self.assertTrue(result)
+        self.assertIn('data', result,)
+
+        # wrongly defined plugins should not load
+        with self.assertRaises(ValueError):
+            self.mixin_wrong.has_api_call()
+
+        # cover wrong token setting
+        with self.assertRaises(ValueError):
+            self.mixin_wrong.has_api_call()
+
+
 class IntegrationPluginBaseTests(TestCase):
     """ Tests for IntegrationPluginBase """