mirror of
				https://github.com/inventree/InvenTree.git
				synced 2025-10-31 05:05:42 +00:00 
			
		
		
		
	readd api call (dont know how that happened)
This commit is contained in:
		| @@ -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 | ||||
| @@ -321,3 +323,107 @@ 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 | ||||
|   | ||||
		Reference in New Issue
	
	Block a user