Source code for nrgpy.api.auth
try:
from nrgpy import logger
except ImportError:
pass
import base64
from datetime import datetime, timedelta
import json
import pickle
import requests
retrieve_token_url = 'https://api.nrgsystems.com/api/RetrieveToken?code=y2/bWG4hRNf1E00lWICOp7nqLvpNPOtaiFf9Wq2bi1iUpdyQdjwicQ=='
data_catalog_url = 'https://api.nrgsystems.com/api/DataCatalog'
convert_url = 'https://api.nrgsystems.com/api/Convert?code=Z6czLero6fQthaM9TZ2DavSN9i7sIeESG/xxGr88JYYoIwypjL/7Uw=='
export_url = 'https://api.nrgsystems.com/api/Export?code=2ZGPXDO8dmHF5IZdm3Qaqrlkf9Gs8930oFeN/MCwX8vcnazvCDkRdg=='
upload_url = "https://api.nrgsystems.com/api/Upload?code=YSy3yEeC6aYMNG9setSKvWe9tZAJJYQtXam1tGT7ADg9FTTCaNqFCw=="
token_file = '.nrgpy_token'
[docs]class nrg_api(object):
"""
Parent class for NRG API functionality
"""
def __init__(self, client_id='', client_secret=''):
logger.debug(f"nrg legacy api init ")
self.client_id = client_id
self.client_secret = client_secret
if self.client_id and self.client_secret:
self.maintain_session_token()
else:
print('[Access error] Valid credentials are required.\nPlease contact support@nrgsystems.com or visit \nhttps://services.nrgsystems.com for API access')
logger.error('[Access error] Valid credentials are required. Please API credentials')
[docs] def request_session_token(self):
"""generates a new session token for convert service api
requires an active account with NRG Systems. to sign
up for an account, go to:
https://services.nrgsystems.com
Parameters
----------
client_id : str
obtained from NRG Systems
client_secret : str
Returns
-------
session_token : str
valid for 24 hour
session_start_time : datetime
start time of 24 hour countdown
"""
logger.debug("session token requested")
print("{} | Requesting session token ... ".format(datetime.now().strftime('%Y-%m-%d %H:%M:%S')), end="", flush=True)
request_token_header = {'content-type': 'application/json'}
request_payload = {'client_id': '{}'.format(self.client_id), 'client_secret': '{}'.format(self.client_secret)}
self.resp = requests.post(data=json.dumps(request_payload), headers=request_token_header, url=retrieve_token_url)
self.session_start_time = datetime.now()
if self.resp.status_code == 200:
logger.debug(f"{self.resp.text}")
print("[OK]")
logger.info("new session token OK")
self.session_token = json.loads(self.resp.text)['access_token']
logger.debug(f"bearer token: {self.session_token}")
else:
logger.error("unable to get session token")
print("[FAILED] | unable to get session token.")
self.session_token = False
[docs] def token_valid(self):
"""check if token is still valid
Parameters
----------
session_start_time : datetime
generated at time of token request
Returns
-------
status : bool
true if still valid, false if expired
"""
if datetime.now() < self.session_start_time + timedelta(hours=18):
if self.session_token is not False:
return True
return False
[docs] def save_token(self, filename=token_file):
"""save session token in token pickle file"""
with open(filename, 'wb') as f:
pickle.dump([self.session_token, self.session_start_time], f)
[docs] def load_token(self, filename=token_file):
"""read session token from pickle file"""
with open(filename, 'rb') as f:
self.session_token, self.session_start_time = pickle.load(f)
[docs] def maintain_session_token(self, filename=token_file):
"""maintain a current/valid session token for data service api"""
try:
self.load_token(filename=token_file)
if not self.token_valid():
self.request_session_token()
self.save_token()
except:
self.request_session_token()
self.save_token()
[docs] def prepare_file_bytes(self, filename=''):
file_bytes = base64.encodebytes(open(filename, 'rb').read())
return file_bytes