We wanted to integrate our Machine Learning platform (we use Dataiku) with QlikSense so that we could refresh a dashboard after the data was loaded by our ML pipeline. Our preference is to use Python since Dataiku supports that natively and most of the ML space is using Python. I looked at the QRS API documentation and I struggled to even get the most simple APIs working. IMHO the QRS API is over complicated, inconsistent and not too well documented. While the documentation covers most of what you need there are often many peculiarities and inconsistencies that are in different places. Some API endpoints return something, some don't, HTTP codes vary between API endpoints and even the ones that return data the data structures are not consistent.
Anyway enough moaning, it took me a while to write this code. I am sharing it here so that someone else don't need to spend as much time as I spent on this.
Test ed on Python 3.6.8, you need these packages installed:
pytz==2020.5
requests==2.22.0
pandas>=1.0,<1.1
ntlm-auth==1.5.0
requests-ntlm==1.1.0
requests-toolbelt==0.9.1
The code returns a Pandas Data Frame with the result of the task you executed. It uses NTLM to authenticate to the QlikSense Server.
import datetime
import random
import string
import json
import requests
import re
import traceback
from time import sleep
from requests_ntlm import HttpNtlmAuth
from requests_toolbelt.utils import dump
# Random Cross Site Forgery key
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Connect-API-Using-Xrfkey-Headers.htm
xrfkey = ''.join(random.choices(string.ascii_uppercase + string.ascii_lowercase + string.digits, k=16))
qliksense_server = 'qliksenser.acme.com'
# Raw string to avoid having to escape the forward slash
ntlm_username = r'domain\user'
ntlm_password = 'password'
session = requests.Session()
# Set internal certificate store to trust internal issued SSL certs
session.verify = '/etc/pki/tls/certs/ca-bundle.trust.crt'
session.auth = HttpNtlmAuth(ntlm_username, ntlm_password)
# User Agent must be set to Windows
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Example-Connect-cURL-Windows.htm
http_headers = {'User-Agent': 'Windows', 'x-qlik-xrfkey': xrfkey}
# Query the /qrs/about API first to obtain the authentication cookie
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Example-Connect-POST-Proxy.htm
service_api = '/qrs/about'
# Uses Python 3 - see https://realpython.com/python-f-strings/
api_endpoint_url = f'https://{qliksense_server}{service_api}?xrfkey={xrfkey}'
try:
print('Authenticate against QlikSense Server')
# Must allow redirects
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Example-Connect-POST-Proxy.htm
print('Calling the following endpoint: ' + api_endpoint_url)
response = session.get(api_endpoint_url, data=None, headers=http_headers, allow_redirects=True, timeout=30)
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('API call returned an error')
content_type = response.headers.get('Content-Type', '')
if 'application/json' in content_type:
try:
response_body = json.dumps(response.json(), indent=2, sort_keys=True)
except json.JSONDecodeError:
response_body = response.text
else:
response_body = response.text
# Expected response code 200 - see https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=611
if response.status_code != 200:
print('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
# Dump and print full response
print(response_body)
response_data = dump.dump_all(response)
print(response_data.decode('utf-8'))
raise ValueError('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
else:
print('Call successful')
try:
print('Check Session Cookie exists')
# Cookie will be passed automatically to the next request as we are using the same requests session object
session_cookie = response.cookies.get_dict()['X-Qlik-Session']
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('Unable to get Session Cookie')
print('Execute a task')
# Query the /qrs/task/ API to execute a task
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Task-Start-By-Name.htm
service_api = '/qrs/task/start/synchronous?name=your%20task%20name'
# Uses Python 3 - see https://realpython.com/python-f-strings/
api_endpoint_url = f'https://{qliksense_server}{service_api}&xrfkey={xrfkey}'
print('Calling the following endpoint: ' + api_endpoint_url)
try:
# Must use POST for this API https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Task-Start-By-Name.htm
response = session.post(api_endpoint_url, data=None, headers=http_headers, allow_redirects=True, timeout=30)
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('API call returned an error')
content_type = response.headers.get('Content-Type', '')
if 'application/json' in content_type:
try:
response_body = json.dumps(response.json(), indent=2, sort_keys=True)
except json.JSONDecodeError:
response_body = response.text
else:
response_body = response.text
# Expected response code 201 - see https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=611
if response.status_code != 201:
print('HTTP Status Code is not 201, got HTTP code ' + str(response.status_code))
# Dump and print full response
print(response_body)
response_data = dump.dump_all(response)
print(response_data.decode('utf-8'))
raise ValueError('HTTP Status Code is not 201, got HTTP code ' + str(response.status_code))
else:
print('Call successful')
try:
print('Get Execution ID')
execution_id = response.json()['value']
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('Unable to parse Execution ID')
# Only proceed if Execution ID is not empty
if execution_id:
print('Check the state of the task execution')
# Query the /qrs/executionsession/ API to check the state of the task execution
# See https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=772
service_api = '/qrs/executionsession/' + execution_id
# Uses Python 3 - see https://realpython.com/python-f-strings/
api_endpoint_url = f'https://{qliksense_server}{service_api}?xrfkey={xrfkey}'
loop_counter = 0
while True:
print('Calling the following endpoint: ' + api_endpoint_url)
try:
# Must use GET for this API https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=772
response = session.get(api_endpoint_url, data=None, headers=http_headers, allow_redirects=True, timeout=30)
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('API call returned an error')
# Expected response code 200 - see https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=772
# Documentation says endpoint should return an empty GUID ( 00000000-0000-0000-0000-000000000000 ) when the task finishes but I get a HTTP 404 with this error: [{"schemaPath":"ExecutionSession","errorText":"Cannot find the item for the \"Get\" operation."}]
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-Task-Start-By-Name.htm
if response.status_code == 200:
print('Call successful, task still running, sleep 15 seconds and continue')
sleep(15)
continue
elif response.status_code == 404:
print('Call successful, task completed, break')
break
else:
print('HTTP Status Code is not 200 nor 404, got HTTP code ' + str(response.status_code))
# Dump and print full response
print(response_body)
response_data = dump.dump_all(response)
print(response_data.decode('utf-8'))
raise ValueError('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
print('Get Task Execution details')
# Query the /qrs/executionsession/ API to check the state of a session
# See https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=772
service_api = '/qrs/executionresult?filter=ExecutionId%20eq%20' + execution_id
# Uses Python 3 - see https://realpython.com/python-f-strings/
api_endpoint_url = f'https://{qliksense_server}{service_api}&xrfkey={xrfkey}'
print('Calling the following endpoint: ' + api_endpoint_url)
try:
# Must use GET for this API https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=772
response = session.get(api_endpoint_url, data=None, headers=http_headers, allow_redirects=True, timeout=30)
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('API call returned an error')
# Expected response code 200 - see https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=772
if response.status_code != 200:
print('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
# Dump and print full response
print(response_body)
response_data = dump.dump_all(response)
print(response_data.decode('utf-8'))
raise ValueError('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
else:
print('Call successful')
print('Get Task Execution details')
content_type = response.headers.get('Content-Type', '')
if 'application/json' in content_type:
try:
task_id = response.json()[0]['id']
task_status_id = response.json()[0]['status']
task_start_time = str(pd.to_datetime(response.json()[0]['startTime']).astimezone('Europe/London').to_pydatetime().strftime('%d-%b-%Y %H:%M:%S'))
task_stop_time = str(pd.to_datetime(response.json()[0]['stopTime']).astimezone('Europe/London').to_pydatetime().strftime('%d-%b-%Y %H:%M:%S'))
task_duration = str(datetime.timedelta(seconds=round(response.json()[0]['duration'] / 1000, 0)))
# TaskExecutionStatus => ExecutionResult.Status
# 0: NeverStarted
# 1: Triggered
# 2: Started
# 3: Queued
# 4: AbortInitiated
# 5: Aborting
# 6: Aborted
# 7: FinishedSuccess
# 8: FinishedFail
# 9: Skipped
# 10: Retry
# 11: Error
# 12: Reset
if task_status_id != 7:
raise ValueError('Task Status ID expect 7 got : ' + str(task_status_id))
else:
print('Task Executed Succesfully')
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('Unable to parse Task Execution details')
else:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('Unable to parse Task Execution details')
print('Get API Enums')
# Query the /qrs/about/api/enums API first to obtain all the valid codes
# See https://help.qlik.com/en-US/sense-developer/August2021/Subsystems/RepositoryServiceAPI/Content/Sense_RepositoryServiceAPI/RepositoryServiceAPI-About-API-Get-Enums.htm
service_api = '/qrs/about/api/enums'
# Uses Python 3 - see https://realpython.com/python-f-strings/
api_endpoint_url = f'https://{qliksense_server}{service_api}?xrfkey={xrfkey}'
print('Calling the following endpoint: ' + api_endpoint_url)
try:
# Must use GET for this API https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=224
response = session.get(api_endpoint_url, data=None, headers=http_headers, allow_redirects=True, timeout=30)
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('API call returned an error')
# Expected response code 200 - see https://help.qlik.com/en-US/sense-developer/August2021/APIs/RepositoryServiceAPI/index.html?page=224
if response.status_code != 200:
print('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
# Dump and print full response
print(response_body)
response_data = dump.dump_all(response)
print(response_data.decode('utf-8'))
raise ValueError('HTTP Status Code is not 200, got HTTP code ' + str(response.status_code))
else:
print('Call successful')
print('Parse API Enums')
content_type = response.headers.get('Content-Type', '')
if 'application/json' in content_type:
try:
task_status_description = re.sub(r'\B([A-Z])', r' \1', response.json()['TaskExecutionStatus']['values'][task_status_id][3:])
except Exception as error:
print('Caught this error: ' + repr(error))
traceback.print_exc()
raise ValueError('Unable to parse API Enums')
print('Write output to DataFrame')
result_list = []
result_list.append([task_id, task_status_id, task_status_description, task_start_time, task_stop_time, task_duration])
df = pd.DataFrame(result_list, columns=['task_id', 'task_status_id', 'task_status_description', 'task_start_time', 'task_stop_time', 'task_duration'])
print('All Done')
