Python Qlik Sense QMC Get Failed / Started / Queued Tasks

Aşağıdaki python kodu ile Qlik Sense QMC üzerindeki hata almış, başlamış ve sıraya girmiş taskları HTML formatında alabilirsiniz.

Burada kullandığım qsAPI kütüphanesini linkten indirip python projenizin içerisine eklemeniz gerekmekte.

# -*- coding: utf-8 -*-
import qsAPI
from datetime import datetime, timedelta
import sys
import io
from datetime import datetime
from tabulate import tabulate

sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding='utf-8')
sys.stderr = io.TextIOWrapper(sys.stderr.detach(), encoding='utf-8')


def connectQliksense(proxy, domain, user, passwd):
    qlik_conn = qsAPI.QRS(proxy=proxy, user=(domain, user, passwd))
    return qlik_conn


# html tablo döndüren fonksiyon
def bordered_html_table(html_data, headers):
    add_border_to_html = "<table border=\"1\">"
    html_table = tabulate(html_data, headers, tablefmt="html")
 
    html_table = html_table.replace("<table>", add_border_to_html)
    return html_table


def convert_qmc_status(status):
    if status == 2:
        status_string = "Started"
    elif status == 3:
        status_string = "Queued"
    elif status == 7:
        status_string = "Success"
    elif status == 8:
        status_string = "Failed"
    else:
        status_string = status
    return status_string


# take third element for sort
def sort_list(elem):
    return elem[4]


# qliksense connection variables
proxy_qliksense_tr = "PROXY SERVER NAME OR IP"
domain_qliksense_tr = "DOMAIN"
user_qliksense_tr = "USERNAME"
passwd_qliksense_tr = "PASSWORD"
# connect to qliksense
qliksense_conn = connectQliksense(proxy_qliksense_tr, domain_qliksense_tr, user_qliksense_tr, passwd_qliksense_tr)
# get app list
tasks = qliksense_conn.TaskGet()

today = datetime.today().date()
task_html = list()
for task in tasks:
    task_list = list()
    # print(task)
    start_time = datetime.strptime(task['operational']['lastExecutionResult']['startTime'][:-5], '%Y-%m-%dT%H:%M:%S') \
                 + timedelta(hours=3)
    if task['operational']['lastExecutionResult']['status'] in [2, 3, 8]:
        task_list.append(task['name'])
        if "app" in task:
            task_list.append(task['app']['stream']['name'])
        else:
            task_list.append("No Stream")
        task_list.append(task['operational']['lastExecutionResult']['executingNodeName'])
        task_list.append(convert_qmc_status(task['operational']['lastExecutionResult']['status']))
        task_list.append(start_time)
        task_list.append(task['operational']['lastExecutionResult']['stopTime'])
        task_list.append(round(task['operational']['lastExecutionResult']['duration'] / 1000 / 60, 2))
        task_html.append(task_list)

# create headers for list
headers = ["Task Name", "Stream Name", "executingNodeName", "status", "startTime", "endTime", "duration"]
# sort list
task_html.sort(reverse=True, key=sort_list)
# create html table for list
table = bordered_html_table(task_html, headers)

print(table)