Welcome to Zuora Product Documentation

Explore our rich library of product information

Sample codes in Python

Sample Python codes to download BI view

The following are the sample codes in Python to download the BI view.

def download(view_name, field_list, page_number, page_size, output_type, client_id, all_pages,
             from_page, to_page, from_date, to_date, retry_interval,
             retry_attempt, use_token, output, max_pages, directory):
    """Download bi view"""
    if page_number is not None:
        from_page = page_number
        to_page = page_number
    o_start = time.time()
    request_id = use_token
    from prettytable import PrettyTable
    x = PrettyTable()
    x.field_names = ["Page Number",
                     "Duration", "Attempt", "Response Code"]
    end_point_host = "https://my-tenant.revprooncloud.com/api"
    for pg in range(from_page, to_page + 1):
        from os import system, name
        p_start = time.time()
        import requests
        token, _ = get_token()
        if request_id is not None:
            headers = {'Token': token,
                       'continuation-token': request_id}
        else:
            headers = {'Token': token}
        params = []
        params.append(make_query_param("clientId", client_id))
        params.append(make_query_param("fromDate", from_date))
        params.append(make_query_param("toDate", to_date))
        params.append(make_query_param("pagenum", pg))
        params.append(make_query_param("pageSize", page_size))
        params.append(make_query_param("outputType", output_type))
        if max_pages is not None:
            params.append(make_query_param("max_pages", max_pages))
        query_string = ""
        for a in params:
            query_string = query_string + a + "&"
        url = end_point_host + \
            '/integration/v2/biviews/{0}?{1}'.format(view_name, query_string)
        import jsonpickle
        field = []
        if field_list is not None:
            field = field_list.split(",")
        post_response = requests.get(url, data=jsonpickle.encode(field), headers=headers)
        attempt = 1
        if post_response.status_code != 200:
            if "Page still not yet cached" in post_response.text:
                for _ in range(1, retry_attempt):
                    attempt = attempt + 1
                    time.sleep(retry_interval)
                    post_response = requests.get(
                        end_point_host + '/integration/v2/biviews/{0}?{1}'.format(view_name, query_string), headers=headers)
                    if post_response.status_code == 200:
                        break
                if post_response.status_code != 400:
                    print("Stopping.")
                    break
        if request_id is None:
            if 'continuation_token' in post_response.headers:
                request_id = post_response.headers.get('continuation_token')
                print("Continuation Token: {0}".format(request_id)) 
        else:
            print('No Token found!')
        p_end = time.time()
        elapsed = round(p_end - p_start)
        if output == "file" and directory is not None:
            write_to_file(directory, pg, post_response.text)
        x.add_row([pg, elapsed, attempt, post_response.status_code])
        print(x)
        print("Continuation Token: {0}".format(request_id))
        if post_response.status_code != 200:
            print("Stopping. Code: {0}".format(post_response.status_code))
            break
    o_end = time.time()
    elapsed = round(o_end - o_start)
    x.add_row(["Total", elapsed, '', ''])
    print(x)
    print("Continuation Token: {0}".format(request_id))