42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from typing import List
|
|
from flask_jwt_extended import get_jwt
|
|
import requests
|
|
|
|
from config import AUTHENTIK_BASEURL
|
|
from .error_handler import AuthentikError
|
|
|
|
class AuthentikApi: # TODO: check if can be replaced with apispec generated api?
|
|
@staticmethod
|
|
def __handleError(res):
|
|
if res.status_code >= 400:
|
|
message = res.json()["error"]["message"]
|
|
raise AuthentikError(message, res.status_code)
|
|
|
|
@staticmethod
|
|
def __token():
|
|
jwt = get_jwt()
|
|
return jwt["sub"]["refresh_token"]
|
|
|
|
@staticmethod
|
|
def get(url):
|
|
try:
|
|
res = requests.get(f"{AUTHENTIK_BASEURL}{url}", headers={
|
|
"Authorization": f"Bearer {AuthentikApi.__token()}"})
|
|
AuthentikApi.__handleError(res)
|
|
if ("pagination" in res.json()):
|
|
return AuthentikApi.__paginate(res)
|
|
return res.json()
|
|
except AuthentikError as err:
|
|
raise err
|
|
except:
|
|
raise AuthentikError()
|
|
|
|
@staticmethod
|
|
def __paginate(res: requests.Response): # TODO: test this
|
|
results = res.json()["results"]
|
|
for page in range(1, res.json()["pagination"]["total_pages"]):
|
|
res = requests.get(
|
|
f"{res.request.url}", headers=res.request.headers, params={'page': page})
|
|
AuthentikApi.__handleError(res)
|
|
results.append(res.json()["results"])
|
|
return results |