How to Connect and Execute SPOTFIRE Rest API Using Python

||
Posted 1 year, 10 months ago
||
Views 266
||
2 min read
0 reactions

Below are the script to connect SPOTFIRE REST API using Python.


Prerequiste:
 


  •     Rest API has to enable in Spotfire Server
  •     Generate/Create CLIENT SECRET KEY & CLIENT_ID to connect using OUTH2 token.

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()


0 reactions

Discussion


Actively looking for JOBS
Joined on April 15, 2020


Latest Videos


Wanna Post Ad?
Reach a large group of audience by posting an ad about your business here. For more details
Drop us a Message