import time
from urllib.parse import urljoin
import requests
from requests import Session
import jwt
class SEAPIClient:
def __init__(
self,
tenant: str,
client_id: str,
client_secret: str,
auth_url: str,
api_identifier: str,
facade_url: str,
timeout: int = 30,
):
self._tenant = tenant
self._client_id = client_id
self._client_secret = client_secret
self._auth_url = auth_url
self._api_identifier = api_identifier
self._facade_url = facade_url
self._session = Session()
self._auth_token = None
self._timeout = timeout
def _get_auth_token(self):
# 5 min expiration buffer
if not self._auth_token or time.time() > self._token_expiry - 300:
self._auth_token = self._fetch_new_auth_token()
return self._auth_token
def _fetch_new_auth_token(self):
url = urljoin(self._auth_url, "/oauth/token")
headers = {"Content-Type": "application/json"}
data = {
"client_id": self._client_id,
"client_secret": self._client_secret,
"audience": self._api_identifier,
"grant_type": "client_credentials",
"tenant": self._tenant,
}
response = requests.post(headers=headers, url=url, json=data, timeout=60)
if response.status_code != 200:
raise Exception(
f"Failed to get auth token: {response.status_code} - {response.text}"
)
token = response.json()["access_token"]
decoded_token = jwt.decode(token, options={"verify_signature": False})
self._token_expiry = decoded_token["exp"]
return f"Bearer {token}"
def _get_url(self):
protocol, base_url = self._facade_url.split("://")
return f"{protocol}://{self._tenant}.{base_url}"
def _create_headers(
self,
content_type: str = "application/json",
version: str = None,
):
headers = {"Content-Type": content_type}
if version:
headers["X-API-Version"] = version
headers["Authorization"] = self._get_auth_token()
return headers
def get(self, endpoint: str, version: str = None):
url = urljoin(self._get_url(), endpoint)
headers = self._create_headers(version=version)
response = self._session.get(headers=headers, url=url, timeout=self._timeout)
return response
def post(self, endpoint: str, data: dict, version: str = None):
url = urljoin(self._get_url(), endpoint)
headers = self._create_headers(version=version)
response = self._session.post(
headers=headers, url=url, json=data, timeout=self._timeout
)
return response
def patch(self, endpoint: str, data: dict, version: str = None):
url = urljoin(self._get_url(), endpoint)
headers = self._create_headers(version=version)
response = self._session.patch(
headers=headers, url=url, json=data, timeout=self._timeout
)
return response
def put(self, endpoint: str, data: dict, version: str = None):
url = urljoin(self._get_url(), endpoint)
headers = self._create_headers(version=version)
response = self._session.put(
headers=headers, url=url, json=data, timeout=self._timeout
)
return response
def delete(self, endpoint: str, version: str = None):
url = urljoin(self._get_url(), endpoint)
headers = self._create_headers(version=version)
response = self._session.delete(headers=headers, url=url, timeout=self._timeout)
return response