Skip to content

Instantly share code, notes, and snippets.

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,
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 = 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 = 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
db2Token=getAuthTokens(credentials["baseURI"], credentials["username"], credentials["pwd"])
print (json.dumps(db2Token, indent=2))
# second parameter is query to process
print (json.dumps(sqlJob, indent=2))
jobres=getSQLJob(credentials["baseURI"],db2Token["token"], sqlJob["id"])
print (json.dumps(jobres, indent=2))
while jobres["status"]=="running":
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.