forked from inventree/InvenTree
-
Notifications
You must be signed in to change notification settings - Fork 4
/
APICallMixin.py
170 lines (133 loc) · 5.7 KB
/
APICallMixin.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
"""Mixin class for making calls to an external API"""
import json as json_pkg
import logging
import requests
from plugin.helpers import MixinNotImplementedError
logger = logging.getLogger('inventree')
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/password (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):
"""Register mixin."""
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):
"""Base url path."""
return f'{self.API_METHOD}://{self.get_setting(self.API_URL_SETTING)}'
@property
def api_headers(self):
"""Returns the default headers for requests with api_call.
Contains a header with the key set in `API_TOKEN` for the plugin it `API_TOKEN_SETTING` is defined.
Check the mixin class docstring for a full example.
"""
headers = {'Content-Type': 'application/json'}
if getattr(self, 'API_TOKEN_SETTING', None):
token = self.get_setting(self.API_TOKEN_SETTING)
if token:
headers[self.API_TOKEN] = token
headers['Authorization'] = f"{self.API_TOKEN} {token}"
return headers
def api_build_url_args(self, arguments: dict) -> str:
"""Returns an encoded path for the provided dict."""
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: str, method: str = 'GET', url_args: dict = None, data=None, json=None, headers: dict = None, simple_response: bool = True, endpoint_is_url: bool = False):
"""Do an API call.
Simplest call example:
```python
self.api_call('hello')
```
Will call the `{base_url}/hello` with a GET request and - if set - the token for this plugin.
Args:
endpoint (str): Path to current endpoint. Either the endpoint or the full or if the flag is set
method (str, optional): HTTP method that should be uses - capitalized. Defaults to 'GET'.
url_args (dict, optional): arguments that should be appended to the url. Defaults to None.
data (Any, optional): Data that should be transmitted in the body - url-encoded. Defaults to None.
json (Any, optional): Data that should be transmitted in the body - must be JSON serializable. Defaults to None.
headers (dict, optional): Headers that should be used for the request. Defaults to self.api_headers.
simple_response (bool, optional): Return the response as JSON. Defaults to True.
endpoint_is_url (bool, optional): The provided endpoint is the full url - do not use self.api_url as base. Defaults to False.
Returns:
Response
"""
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:
if endpoint.startswith('/'):
endpoint = endpoint[1:]
url = f'{self.api_url}/{endpoint}'
# build kwargs for call
kwargs = {
'url': url,
'headers': headers,
}
if data and json:
raise ValueError('You can either pass `data` or `json` to this function.')
if json:
kwargs['data'] = json_pkg.dumps(json)
if data:
kwargs['data'] = data
# run command
response = requests.request(method, **kwargs)
# return
if simple_response:
return response.json()
return response