Simple Python-based test of the Db2 on Cloud REST API

db2APItest.py

# Simple test of the Db2 on Cloud REST API
# Written by Henrik Loeser, hloeser@de.ibm.com
# https://cloud.ibm.com/apidocs/db2-on-cloud
import requests, json, sys, time

# Read credentials from file
def readCreds(filename):
    with open(filename) as data_file:
        credentials = json.load(data_file)
    return credentials

# Generate authentication token from username/password  
def getAuthTokens(baseURI, username, pwd):
    url     = baseURI+"/auth/tokens"
    data    = {"userid": username, "password":pwd}
    response  = requests.post( url, json=data )
    return response.json()

def sendSQLJob(baseURI, token, commands):
    url     = baseURI+"/sql_jobs"
    headers = { "Authorization" : "Bearer "+token, "accept": "application/json" }
    data    = {"commands": commands, "limit":1000, "separator":";", "stop_on_error":"no"}
    response  = requests.post( url, json=data, headers=headers )
    return response.json()

def getSQLJob(baseURI, token, jobid):
    url = baseURI+"/sql_jobs/"+jobid
    headers = { "authorization" : "Bearer "+token, "accept": "application/json" }
    response = requests.get(url, headers=headers)
    return response.json()

if __name__== "__main__":
    # First parameter is credentials file
    credentials=readCreds(sys.argv[1])

    db2Token=getAuthTokens(credentials["baseURI"], credentials["username"], credentials["pwd"])
    print (json.dumps(db2Token, indent=2))
    
    # second parameter is query to process
    sqlJob=sendSQLJob(credentials["baseURI"],db2Token["token"],sys.argv[2])
    print (json.dumps(sqlJob, indent=2))

    jobres=getSQLJob(credentials["baseURI"],db2Token["token"], sqlJob["id"])
    print (json.dumps(jobres, indent=2))

    while jobres["status"]=="running":
        time.sleep(10)
        jobres=getSQLJob(credentials["baseURI"],db2Token["token"], sqlJob["id"])
        print (json.dumps(jobres, indent=2))