class RefinitivConnection:
def __init__(self, username, password, appid, session):
self.username = username
self.password = password
self.appid = appid
self.session = session
self.token = self.getToken()
# Send HTTP request for all services, raise error if unsuccessful
def sendRequest(self, url, requestMsg=None):
result = requests.post(
url=url,
data=json.dumps(requestMsg),
headers=self.headers,
proxies={
'http': self.session.proxies["https"],
'https': self.session.proxies["http"]
}
)
if result.status_code != 200:
print_centre_hashed('Request fail: response status %s'%(result.status_code))
if result.status_code == 500:
print('Error: %s' % (json.dumps(result.json(),sort_keys=True, indent=2, separators=(',', ':'))))
result.raise_for_status()
return result
# Authenticates to the Refinitiv API returning the token, updates headers with new token/appid info
def getToken(self):
token = None
authUrl = 'https://api.rkd.refinitiv.com/api/TokenManagement/TokenManagement.svc/REST/Anonymous/TokenManagement_1/CreateServiceToken_1'
self.headers = {'content-type': 'application/json;charset=utf-8'}
authMsg = {
'CreateServiceToken_Request_1': {
'ApplicationID':self.appid,
'Username':self.username,
'Password':self.password
}
}
print_centre_hashed('Sending Token Request')
authResult = self.sendRequest(authUrl, authMsg)
if authResult and authResult.status_code == 200:
print_centre_hashed('Authen success - Token Generated')
token = authResult.json()['CreateServiceToken_Response_1']['Token']
self.headers = {
'content-type': 'application/json;charset=utf-8',
'X-Trkd-Auth-ApplicationID': self.appid,
'X-Trkd-Auth-Token': token
}
return token