import time from typing import Union, Any import requests from flask import session, current_app class Delphi: TOKEN_KEY = 'delphi_api_session_token' TOKEN_KEY_EXPIRES_AT = 'delphi_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['DELPHI_API_ID'], 'client_secret': current_app.config['DELPHI_API_SECRET'], 'grant_type': 'client_credentials' } i = 0 r = None while (not hasattr(r, 'status_code') or r.status_code != 200) and i < 3: time.sleep(i) print(f"DELPHI!!!!! {i}") r = requests.post(current_app.config['DELPHI_API_URL'] + 'oauth/token', json=payload) i += 1 r = r.json() session[cls.TOKEN_KEY] = r['access_token'] session[cls.TOKEN_KEY_EXPIRES_AT] = r['expires_at'] 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['DELPHI_API_URL'] + endpoint headers = {"Authorization": "Bearer " + token} r = requests.get(endpoint, params=payload or {}, headers=headers) return r.json(), r.url