112 lines
3.6 KiB
Plaintext
112 lines
3.6 KiB
Plaintext
from typing import Dict, Optional, final
|
|
|
|
import requests
|
|
import xmltodict
|
|
from rest_framework import status
|
|
|
|
|
|
class BaseAPI:
|
|
__api_name__ = "BaseAPI"
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
def get_host(self) -> str:
|
|
raise NotImplementedError("Subclasses must implement _get_host() method")
|
|
|
|
def build_headers(self) -> Dict[str, str]:
|
|
return {}
|
|
|
|
def build_params(self) -> Dict[str, str]:
|
|
return {}
|
|
|
|
def repair_authentication(self) -> bool:
|
|
"""Callback to repair authentication, e.g. refresh token or re-login."""
|
|
return False
|
|
|
|
@final
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
data: Optional[Dict | str | bytes] = None,
|
|
json: Optional[Dict] = None,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
params: Optional[Dict[str, str]] = None,
|
|
custom_host: Optional[str] = None,
|
|
timeout: int = 30,
|
|
use_build_headers: bool = True,
|
|
) -> Dict:
|
|
host = custom_host or self.get_host()
|
|
url = f"{host}{path}"
|
|
combined_headers: Optional[Dict[str, str]] = {
|
|
**(headers or {}),
|
|
}
|
|
if use_build_headers:
|
|
combined_headers = {
|
|
**self.build_headers(),
|
|
**(headers or {}),
|
|
}
|
|
combined_params: Optional[Dict[str, str]] = {
|
|
**self.build_params(),
|
|
**(params or {}),
|
|
}
|
|
|
|
if combined_headers == {}:
|
|
combined_headers = None
|
|
if combined_params == {}:
|
|
combined_params = None
|
|
|
|
try:
|
|
response = requests.request(
|
|
method=method,
|
|
url=url,
|
|
headers=combined_headers,
|
|
params=combined_params,
|
|
data=data,
|
|
json=json,
|
|
timeout=timeout,
|
|
)
|
|
if (
|
|
response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
or response.status_code == status.HTTP_403_FORBIDDEN
|
|
):
|
|
if self.repair_authentication():
|
|
host = custom_host or self.get_host()
|
|
url = f"{host}{path}"
|
|
combined_headers: Optional[Dict[str, str]] = {
|
|
**self.build_headers(),
|
|
**(headers or {}),
|
|
}
|
|
combined_params: Optional[Dict[str, str]] = {
|
|
**self.build_params(),
|
|
**(params or {}),
|
|
}
|
|
response = requests.request(
|
|
method=method,
|
|
url=url,
|
|
headers=combined_headers,
|
|
params=combined_params,
|
|
data=data,
|
|
timeout=timeout,
|
|
)
|
|
response.raise_for_status()
|
|
if response.status_code == status.HTTP_204_NO_CONTENT:
|
|
return {}
|
|
|
|
content_type = response.headers.get("Content-Type", "")
|
|
|
|
if "xml" in content_type:
|
|
return xmltodict.parse(response.text)
|
|
if "application/json" in content_type:
|
|
return response.json()
|
|
else:
|
|
raise ValueError(
|
|
"Unsupported Content-Type: {content_type} must be application/json"
|
|
)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
raise e
|
|
except ValueError as e:
|
|
raise e
|
|
|