Below are the script to connect SPOTFIRE REST API using Python.
Prerequiste:
This is just simple template script. Using this skelton , we can extend our logic.
import requests
import ssl
import time
import datetime
import logging
import os
ssl._create_default_https_context = ssl._create_unverified_context
logging.getLogger().setLevel(logging.INFO)
#Login into Spotfire API
def login(base_url):
print("Getting token...")
login_time = time.time()
print(datetime.datetime.fromtimestamp(login_time).strftime('%Y-%m-%d %H:%M:%S'))
header_gs = {
'accept': 'application/json',
# Using Base Encode64 logic get the below Authorization id(clientid:clientsecret)
'Authorization': 'Basic {TOKEN}',
'content-type': 'application/x-www-form-urlencoded'
}
data_get = {
'grant_type': 'client_credentials',
"scope": "api.rest.automation-services-job.execute"
}
try:
r = requests.post(base_url+'/spotfire/oauth2/token', headers=header_gs, data=data_get)
if r.status_code == 200:
auth_token = r.json()
logging.info('Connected Successfully')
print(auth_token['access_token'])
return auth_token['access_token']
except Exception as e:
# print('err')
logging.info(f'{e}')
exit(0)
# Execute the job & get the job id
def start_library(base_url,token):
header_gs={
'authorization': 'Bearer '+str(token)
}
data_get = {
'id': 'dce27525-f4c9-43d5-a2c2-9b2d90622ddf'
}
try:
r = requests.post(base_url+'/spotfire/api/rest/as/job/start-library', headers=header_gs, data=data_get)
# print(r.json()['JobId'])
logging.info('Executed the Job')
return r.json()['JobId']
except Exception as e:
logging.info(f'{e}')
exit(0)
#Get the status of the job
def get_status(base_url,token,jobid):
header_gs = {
'authorization': 'Bearer ' + str(token)
}
try:
r = requests.get(base_url+f'/spotfire/api/rest/as/job/status/{jobid}', headers=header_gs)
# print(r.json())
logging.info('Job ID has been generated and passed to final check status function')
return r.json()['StatusCode']
except Exception as e:
logging.info(f'{e}')
exit(0)
client_id = 'token.oauth-clients.spotfire.tibco.com'
client_secret = 'SECRET_KEY'
base_url = 'SPOTFIRE URL'
# Exeucte all above functions
def final_status():
get_token = login(base_url)
get_job_id = start_library(base_url, get_token)
get_status(base_url,get_token, get_job_id)
time.sleep(60)
check_finish_status = get_status(base_url,get_token, get_job_id)
# print(check_finish_status)
if check_finish_status == 'Finished':
logging.info('Job has been completed successfully')
return 'Success'
exit(0)
requests.
if __name__ == '__main__':
final_status()