Skip to main content
Announcements
Qlik Connect 2024! Seize endless possibilities! LEARN MORE
cancel
Showing results for 
Search instead for 
Did you mean: 
Not applicable

Connexion between python and Qlik Sense

Hello,

I would like to load data QlikSense a server application via a Python application. I saw in the documentation that we can do it via websocket, and I cannot perform this operation.

below the python source code

import websocket

import json

  1. websocket.create_connection = socket ('ws: // [ma_machine]:4747')
  2. json.dumps data = ({'jsonrpc': '2.0', 'id': 3, 'method', 'DoReload', 'handle': 1, 'params': {'qMode': 0, 'qPartial': false, 'qDebug'. false}}) encode ('utf-8')
  3. socket.send (data)

Thanking you in advance.

1 Reply
tts
Employee
Employee

Hi, this is a sample Python script which uses websocket-client package.

Note: I'm using Python 3.5.1

----------

import websocket

import ssl

import json

import threading

from time import sleep

ws = None

current_id = 0

QES_return = None

def wait_next(id):

    global current_id

    while current_id == id:

        sleep(0.5)

def on_open(ws):

    #print("### open ###")

    global current_id

    current_id += 1

def on_message(ws, message):

    #print("### message ###")

    #print(message)

    global QES_return

    QES_return = json.loads(message)['result']['qReturn']

    global current_id

    current_id += 1

def on_error(ws, error):

    #print("### error ###")

    print(error)

def on_close(ws):

    print("### closed ###")

# worker thread

def ws_run():

    global ws

    ws = websocket.WebSocketApp("wss://qesXXXhostname:4747", on_message = on_message, on_error = on_error, on_close = on_close, header = {'X-Qlik-User: UserDirectory=QESXXXHOSTNAME; UserId=administrator'})

    ws.on_open = on_open

    ssl_option = ({"ca_certs": "root.pem",

                   "certfile": "client.pem",

                   "keyfile": "client_key.pem",

                   "cert_reqs": ssl.CERT_REQUIRED,

                   "server_side": False})

    ws.run_forever(sslopt=ssl_option)

# main thread

#websocket.enableTrace(True)

wst = threading.Thread(target=ws_run)

wst.daemon = True

wst.start()

# wait for on_open

wait_next(current_id)

py_obj = {'jsonrpc': '2.0', 'id': current_id, 'method': 'OpenDoc', 'handle': -1, 'params': ['d366fcd7-92c3-46c3-8fcf-39f2ab7f5205']}

json_str = json.dumps(py_obj)

ws.send(json_str)

wait_next(current_id)

qHandle = QES_return['qHandle']

#print(qHandle)

py_obj = {'jsonrpc': '2.0', 'id': current_id, 'method': 'DoReload', 'handle': qHandle, 'params': []}

json_str = json.dumps(py_obj)

ws.send(json_str)

wait_next(current_id)

#print(QES_return)

----------