65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
import requests
|
|
|
|
class ApiClient():
|
|
def __init__(self, api_url, default_token=None):
|
|
super().__init__()
|
|
self.api_url = api_url
|
|
self.default_token = default_token
|
|
|
|
def __request(self, endpoint, payload=None, params=None, method='GET', token=None):
|
|
if payload is None:
|
|
payload = {}
|
|
if params is None:
|
|
params = {}
|
|
|
|
request_url = f'{self.api_url}{endpoint}'
|
|
|
|
headers = {}
|
|
if token is None:
|
|
token = self.default_token
|
|
if token:
|
|
headers['Authorization'] = f'Bearer {token}'
|
|
try:
|
|
if method.upper() == 'POST':
|
|
response = requests.post(request_url, json=payload, headers=headers)
|
|
elif method.upper() == 'GET':
|
|
response = requests.get(request_url, params=params, headers=headers)
|
|
elif method.upper() == 'PUT':
|
|
response = requests.put(request_url, json=payload, headers=headers)
|
|
elif method.upper() == 'DELETE':
|
|
response = requests.delete(request_url, json=payload, headers=headers)
|
|
else:
|
|
raise ValueError("Unsupported HTTP method: {}".format(method))
|
|
|
|
if response.status_code != 200:
|
|
self.warning(f'Return code {response.status_code} from {request_url} with request {method} with message {response.text}')
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
self.error("An error occurred:", e)
|
|
return None
|
|
|
|
return response
|
|
|
|
def get(self, endpoint, params=None, token=None):
|
|
return self.__request(endpoint, method='GET', params=params, token=token)
|
|
|
|
def post(self, endpoint, payload=None, token=None):
|
|
return self.__request(endpoint, payload=payload, method='POST', token=token)
|
|
|
|
def put(self, endpoint, payload=None, token=None):
|
|
return self.__request(endpoint, payload=payload, method='PUT', token=token)
|
|
|
|
def delete(self, endpoint, payload=None, token=None):
|
|
return self.__request(endpoint, payload=payload, method='DELETE', token=token)
|
|
|
|
def reassign_token(self, token):
|
|
self.default_token = token
|
|
|
|
def warning(self, message):
|
|
print(f"WARNING: {message}")
|
|
|
|
def error(self, message, exception=None):
|
|
print(f"ERROR: {message}")
|
|
if exception:
|
|
print(f"Exception: {exception}")
|