Coverage for src/backend/InvenTree/plugin/base/integration/APICallMixin.py: 66%

1"""Mixin class for making calls to an external API.""" 

2 

3import json as json_pkg 

4from collections.abc import Iterable 

5from typing import Optional 

6 

7import requests 

8import structlog 

9 

10from plugin import PluginMixinEnum 

11from plugin.helpers import MixinNotImplementedError 

12 

13logger = structlog.get_logger('inventree') 

14 

15 

16class APICallMixin: 

17 """Mixin that enables easier API calls for a plugin. 

18 

19 Steps to set up: 

20 1. Add this mixin before (left of) SettingsMixin and PluginBase 

21 2. Add two settings for the required url and token/password (use `SettingsMixin`) 

22 3. Save the references to keys of the settings in `API_URL_SETTING` and `API_TOKEN_SETTING` 

23 4. (Optional) Set `API_TOKEN` to the name required for the token by the external API - Defaults to `Bearer` 

24 5. (Optional) Override the `api_url` property method if the setting needs to be extended 

25 6. (Optional) Override `api_headers` to add extra headers (by default the token and Content-Type are contained) 

26 7. Access the API in you plugin code via `api_call` 

27 

28 Example: 

29 ``` 

30 from plugin import InvenTreePlugin 

31 from plugin.mixins import APICallMixin, SettingsMixin 

32 

33 

34 class SampleApiCallerPlugin(APICallMixin, SettingsMixin, InvenTreePlugin): 

35 ''' 

36 A small api call sample 

37 ''' 

38 NAME = "Sample API Caller" 

39 

40 SETTINGS = { 

41 'API_TOKEN': { 

42 'name': 'API Token', 

43 'protected': True, 

44 }, 

45 'API_URL': { 

46 'name': 'External URL', 

47 'description': 'Where is your API located?', 

48 'default': 'reqres.in', 

49 }, 

50 } 

51 API_URL_SETTING = 'API_URL' 

52 API_TOKEN_SETTING = 'API_TOKEN' 

53 

54 def get_external_url(self): 

55 ''' 

56 returns data from the sample endpoint 

57 ''' 

58 return self.api_call('api/users/2') 

59 ``` 

60 """ 

61 

62 API_METHOD = 'https' 

63 API_URL_SETTING = None 

64 API_TOKEN_SETTING = None 

65 

66 API_TOKEN = 'Bearer' 

67 

68 class MixinMeta: 

69 """Meta options for this mixin.""" 

70 

71 MIXIN_NAME = 'API calls' 

72 

73 def __init__(self): 

74 """Register mixin.""" 

75 super().__init__() 

76 self.add_mixin(PluginMixinEnum.API_CALL, 'has_api_call', __class__) 

77 

78 @property 

79 def has_api_call(self): 

80 """Is the mixin ready to call external APIs?""" 

81 if not bool(self.API_URL_SETTING): 

82 raise MixinNotImplementedError('API_URL_SETTING must be defined') 

83 if not bool(self.API_TOKEN_SETTING): 

84 raise MixinNotImplementedError('API_TOKEN_SETTING must be defined') 

85 return True 

86 

87 @property 

88 def api_url(self): 

89 """Base url path.""" 

90 return f'{self.API_METHOD}://{self.get_setting(self.API_URL_SETTING)}' 

91 

92 @property 

93 def api_headers(self): 

94 """Returns the default headers for requests with api_call. 

95 

96 Contains a header with the key set in `API_TOKEN` for the plugin it `API_TOKEN_SETTING` is defined. 

97 Check the mixin class docstring for a full example. 

98 """ 

99 headers = {'Content-Type': 'application/json'} 

100 if getattr(self, 'API_TOKEN_SETTING', None): 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 token = self.get_setting(self.API_TOKEN_SETTING) 

102 

103 if token: 

104 headers[self.API_TOKEN] = token 

105 headers['Authorization'] = f'{self.API_TOKEN} {token}' 

106 

107 return headers 

108 

109 def api_build_url_args(self, arguments: dict) -> str: 

110 """Returns an encoded path for the provided dict.""" 

111 groups = [] 

112 for key, val in arguments.items(): 

113 if isinstance(val, Iterable) and not isinstance(val, str): 113 ↛ 115line 113 didn't jump to line 115 because the condition on line 113 was always true

114 val = ','.join([str(a) for a in val]) 

115 groups.append(f'{key}={val}') 

116 return f'?{"&".join(groups)}' 

117 

118 def api_call( 

119 self, 

120 endpoint: str, 

121 method: str = 'GET', 

122 url_args: Optional[dict] = None, 

123 data=None, 

124 json=None, 

125 headers: Optional[dict] = None, 

126 simple_response: bool = True, 

127 endpoint_is_url: bool = False, 

128 **kwargs, 

129 ): 

130 """Do an API call. 

131 

132 Simplest call example: 

133 ```python 

134 self.api_call('hello') 

135 ``` 

136 Will call the `{base_url}/hello` with a GET request and - if set - the token for this plugin. 

137 

138 Args: 

139 endpoint (str): Path to current endpoint. Either the endpoint or the full or if the flag is set 

140 method (str, optional): HTTP method that should be uses - capitalized. Defaults to 'GET'. 

141 url_args (dict, optional): arguments that should be appended to the url. Defaults to None. 

142 data (Any, optional): Data that should be transmitted in the body - url-encoded. Defaults to None. 

143 json (Any, optional): Data that should be transmitted in the body - must be JSON serializable. Defaults to None. 

144 headers (dict, optional): Headers that should be used for the request. Defaults to self.api_headers. 

145 simple_response (bool, optional): Return the response as JSON. Defaults to True. 

146 endpoint_is_url (bool, optional): The provided endpoint is the full url - do not use self.api_url as base. Defaults to False. 

147 

148 Returns: 

149 Response 

150 """ 

151 if url_args: 151 ↛ 154line 151 didn't jump to line 154 because the condition on line 151 was always true

152 endpoint += self.api_build_url_args(url_args) 

153 

154 if headers is None: 154 ↛ 157line 154 didn't jump to line 157 because the condition on line 154 was always true

155 headers = self.api_headers 

156 

157 if endpoint_is_url: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true

158 url = endpoint 

159 else: 

160 if endpoint.startswith('/'): 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true

161 endpoint = endpoint[1:] 

162 

163 url = f'{self.api_url}/{endpoint}' 

164 

165 # build kwargs for call 

166 kwargs.update({'headers': headers}) 

167 kwargs.pop('url', None) 

168 

169 if data and json: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true

170 raise ValueError('You can either pass `data` or `json` to this function.') 

171 

172 if json: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true

173 kwargs['data'] = json_pkg.dumps(json) 

174 

175 if data: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true

176 kwargs['data'] = data 

177 

178 # run command 

179 response = requests.request(method, url=url, **kwargs) 

180 

181 # return 

182 if simple_response: 182 ↛ 183line 182 didn't jump to line 183 because the condition on line 182 was never true

183 return response.json() 

184 return response