import time from typing import Union, Any import requests from flask import session, current_app class Apollo: TOKEN_KEY = 'apollo_api_session_token' TOKEN_KEY_EXPIRES_AT = 'apollo_api_session_token_expires_at' @classmethod def get_token(cls): if cls.TOKEN_KEY not in session or cls.TOKEN_KEY_EXPIRES_AT not in session or session[cls.TOKEN_KEY_EXPIRES_AT] <= time.time(): payload = { 'client_id': current_app.config['APOLLO_API_ID'], 'client_secret': current_app.config['APOLLO_API_SECRET'], 'grant_type': 'client_credentials', "audience": "https://apollo-private-api/" } i = 0 r = None while (not hasattr(r, 'status_code') or r.status_code != 200) and i < 3: time.sleep(i) print(f"Apollo!!!!! {i}") r = requests.post("https://prod-dna-apps.auth0.com/" + "oauth/token", json=payload) i += 1 r = r.json() session[cls.TOKEN_KEY] = r['access_token'] session[cls.TOKEN_KEY_EXPIRES_AT] = r['expires_in'] return session[cls.TOKEN_KEY] @classmethod def request(cls, endpoint: str, payload: dict = None, token: str = None) -> Union[Any, str]: from app import application with application.app_context(): if not token: token = cls.get_token() if not endpoint.startswith("http"): endpoint = current_app.config['APOLLO_API_URL'] + endpoint headers = {"Authorization": "Bearer " + token} r = requests.get(endpoint, params=payload or {}, headers=headers) return r.json(), r.url