Created
August 12, 2019 10:03
-
-
Save data-henrik/e5e7d3f197476a4d0a76dec80bf4f77b to your computer and use it in GitHub Desktop.
Simple Python-based test of the Db2 on Cloud REST API
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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