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
