in src/data_load/utils.py [0:0]
def refresh(cls):
"""
Refresh token and save them into class.
"""
logger.info("Refreshing token.")
for i in range(cls._retries):
# try several times if there any error
try:
if environ.get('REFRESH_TOKEN') is not None:
resp = cls.refresh_request(cls._token_endpoint, environ.get('REFRESH_TOKEN'), cls._client_id, cls._client_secret)
else:
resp = cls.login_with_service_principal_creds(cls._token_endpoint, cls._client_id, cls._client_secret)
if 'access_token' in resp:
break
except HTTPError:
if i == cls._retries - 1:
# too many errors, raise original exception
raise
cls._access_token = resp["access_token"]
cls.expire_date = datetime.now().timestamp() + int(resp["expires_in"])
logger.info("Token is refreshed.")