79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
import os
|
|
import time
|
|
|
|
import requests
|
|
|
|
|
|
class BaseAPI:
|
|
CONNECTION_TEST_INTERVAL = 10
|
|
CONNECTION_TEST_ATTEMPTS = 10
|
|
|
|
def __init__(self, app_domain, **kwargs):
|
|
self.app_domain = app_domain
|
|
self.token = None
|
|
self.api_tenant_login_url = f"https://tenant-login-api.{self.app_domain}"
|
|
self.url = None
|
|
self.params = {**kwargs}
|
|
|
|
def __get_connection_test_url(self):
|
|
return f"{self.url}/check_connection"
|
|
|
|
def wait_for_start(self, url=None):
|
|
i = 0
|
|
if not url:
|
|
url = self.__get_connection_test_url()
|
|
for i in range(self.CONNECTION_TEST_ATTEMPTS):
|
|
try:
|
|
self._get(url)
|
|
except Exception:
|
|
time.sleep(self.CONNECTION_TEST_INTERVAL)
|
|
continue
|
|
break
|
|
if i == self.CONNECTION_TEST_ATTEMPTS - 1:
|
|
raise Exception(f"Connection to {url} failed.")
|
|
|
|
def _headers(self, content_type='application/json'):
|
|
headers = {'Content-Type': content_type}
|
|
if self.token:
|
|
headers['Authorization'] = self.token
|
|
return headers
|
|
|
|
def _get(self, endpoint):
|
|
try:
|
|
response = requests.get(endpoint, headers=self._headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
raise requests.exceptions.HTTPError(e.response.text)
|
|
|
|
def _post(self, endpoint, payload):
|
|
try:
|
|
response = requests.post(endpoint, json=payload, headers=self._headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
raise requests.exceptions.HTTPError(e.response.text)
|
|
|
|
def _put(self, endpoint, payload):
|
|
try:
|
|
response = requests.put(endpoint, json=payload, headers=self._headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
raise requests.exceptions.HTTPError(e.response.text)
|
|
|
|
def _delete(self, endpoint):
|
|
response = requests.delete(endpoint, headers=self._headers())
|
|
response.raise_for_status()
|
|
return response.status_code
|
|
|
|
def _upload_file(self, endpoint, file_path, params=None):
|
|
try:
|
|
with open(file_path, 'rb') as file:
|
|
files = {'file': (os.path.basename(file_path), file, 'application/octet-stream')}
|
|
response = requests.post(endpoint, files=files, data=params, headers=self._headers(content_type=None))
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
raise requests.exceptions.HTTPError(e.response.text)
|