Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Simple Python-based test of the Db2 on Cloud REST API
# 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))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.