Skip to main content

Sample codes in Python

Zuora

Sample codes in Python

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))