Sample Python client

A basic example of a Python client for Analytics Reports API

Below you will find an example project demonstrating a minimal python client for the Analytics Reports API. This project does not offer a full API client; it demonstrates the essential interaction with the API using Python.


  • Know your Khoros company key and your Analytics API username and password. See Authentication and API URLs for details
  • Have python 3.x installed
  • Install the 'requests' library: sudo pip install requests && pip install pandas
  • Make the '' file executable: chmod u+x && chmod u+x

Upon executing the script contained herein, the script will:

  • Check your authentication credentials, and if successful ...
  • Request execution of the 'Incoming Posts' report, and if successful ...
  • Block until the report is complete, downloading it to a temporary file

Upon executing the script contained herein, the script will:

  • Ask for the JSON file you want to convert into CSV
  • Save the file converted in the same folder used to save the files downloaded

Detailed execution options are obtained by running ./ -h.

from client import AnalyticsReportClient
from options import AnalyticsReportApiOptions

def main():
    This 'hello, world' demo checks your credentials for the Khoros Analytics reporting API, requests an
    'Incoming Post' report, blocks until that report has completed, then downloads the report result to a
     temporary file.
    opts = AnalyticsReportApiOptions()
    client = AnalyticsReportClient(opts)
    if client.check_auth():
            status_url = client.request_incoming_posts_for_last_24()
            (is_unblock, download_url) = client.block_until_done(status_url)
            if is_unblock:

if __name__ == '__main__':
import os
import tempfile
import time

import requests

class AnalyticsReportClient:
    The goal of this minimal client for the Khoros Analytics reporting API is to demonstrate how a python program might
     interact with the API.

    def __init__(self, opts):
        self.opts = opts

    def check_auth(self):
        print("checking auth for report user {} on company {}".format(self.opts.auth.username, self.opts.company_key))
        auth_check_response = requests.get(
            '{}/api/public/reports/authcheck?companyKey={}'.format(self.opts.url_base, self.opts.company_key),
        if auth_check_response.status_code is not 200:
            print("authentication check failed: {} -- {}".format(auth_check_response.status_code, auth_check_response.json()['result']))
            return False
            print("authentication check passed")
            return True

    def request_incoming_posts_for_last_24(self):
        current_time_millis = int(round(time.time() * 1000))
        millis_in_a_day = 86400000

        start = current_time_millis - millis_in_a_day
        end = current_time_millis

        request_url = "{}/api/public/reports/report/incoming_post?companyKey={}&startTime={}&endTime={}&reportFormat={}" \
            .format(self.opts.url_base, self.opts.company_key, start, end, self.opts.format)
        print("requesting incoming posts report: \n\t{}".format(request_url))

        request_report_response =, auth=self.opts.auth)
        if request_report_response.status_code is not 200:
            print("report request failed: {} -- {}".format(request_report_response.status_code, request_report_response.json()))
            return None
            print("report request successful")
            return request_report_response.json()['result']['statusUrl']

    def block_until_done(self, report_status_url):
        print("blocking until report done: \n\t{}".format(report_status_url))
        while True:
            report_status_response = requests.get(report_status_url, auth=self.opts.auth)
            if report_status_response.status_code is not 200:
                print("report status check failed: {} -- {}".format(report_status_response.status_code, report_status_response.json()))
                return False, ""
            elif report_status_response.json()['result']['result']['runnerState'] == 'CLOSED' \
                    and  report_status_response.json()['result']["result"]["detail"] != 'COMPLETED':
                print("report encountered an error and cannot be downloaded. Detail is {}".format(report_status_response.json()['result']['result'][
                return False, ""
            elif report_status_response.json()['result']['result']['runnerState'] == 'CLOSED' \
                    and  report_status_response.json()['result']["result"]["detail"] == 'COMPLETED':
                print("report status is {} ... could download now".format(report_status_response.json()['result']['result'][
                return True, report_status_response.json()['result']['jobInfo']['downloadUrl']
                print("report status is {} ... waiting".format(report_status_response.json()['result']['result']['runnerState']))

    def download_report_to_tmp_file(self, report_download_url):
        report_download_response = requests.get(report_download_url, auth=self.opts.auth)
        if report_download_response.status_code is not 200:
            print("failed to download report: {} -- {}".format(report_download_response.status_code, report_download_response.json()))
            downloads_dir = "./downloads/"
            os.makedirs(downloads_dir, exist_ok=True)

            fd, path = tempfile.mkstemp(suffix=".{}".format(self.opts.format), prefix="incoming_posts_", dir=downloads_dir)
            os.write(fd, report_download_response.content)
            print("downloaded report and saved to temp file: \n\t{}".format(path))
import argparse
from requests.auth import HTTPBasicAuth

class AnalyticsReportApiOptions:
    Defines options relevant for interacting with the Khoros Analytics reporting API.

    def __init__(self):

        def resolve_base_url(region):
            if region.casefold() == "us-west-2":
                return ""
            elif region.casefold() == "eu-west-1":
                return ""
                return None

        parser = argparse.ArgumentParser(description="A minimal 'hello, world' client demonstrating basic interaction"
                                                     " with the Khoros Analytics reporting API.")

        host_opts = parser.add_argument_group('Host Options', 'Describes which environment you want to exercise.')
        host_opts.add_argument('-r', '--region', default='us-west-2',  required=False,
                               choices=['us-west-2', 'US-WEST-2', 'eu-west-1', 'EU-WEST-1'],
                               help='Valid values are "us-west-2" & "eu-west-1".')

        format_opts = parser.add_argument_group('Format Options', 'Describes your export format.')
        format_opts.add_argument('-f', '--format', default='csv', required=False, choices=['csv', 'CSV', 'json', 'JSON'],
                               help='Valid values are "csv" & "json".')

        auth_opts = parser.add_argument_group('Authentication Options', 'Describes your API credentials.')
        auth_opts.add_argument('-c', '--company', required=True, help='Your company key.')
        auth_opts.add_argument('-u', '--user', required=True, help='Basic-auth user for calling reporting API.')
        auth_opts.add_argument('-p', '--password', required=True, help='Basic-auth password for calling reporting API.')

        parsed = parser.parse_args()

        self.company_key =
        self.auth = HTTPBasicAuth(parsed.user, parsed.password)
        self.url_base = resolve_base_url(parsed.region)
        self.format = parsed.format

        print("Parsed options for exercising Khoros Analytics reporting API; base url {}".format(self.url_base))
from optionsConvertFile import OptionsConvertFile
import pandas as pd
import os

def main():
    Sometimes, Khoros adds columns to exports. By downloading the export in JSON format and preparing your scripts to
    only process the columns you know/care about, the appearance of new columns will not break your processing.
    opts = OptionsConvertFile()

    if os.path.exists('./downloads/{}'.format(opts.fileName)):
        df = pd.read_json (r'./downloads/{}'.format(opts.fileName))
        df.to_csv (r'./downloads/{}.csv'.format(opts.fileName.split('.')[0]), index = None, quoting = 1,
                   columns = ['conversationId', 'conversationWorkQueue', 'conversationPriority',
                              'postProvider', 'externalPostId', 'datePostPublished',
                              'datePostReceived', 'authorHandle', 'externalAuthorHandleId',
                              'allAuthorTags', 'postType', 'postContent', 'postOrder',
                              'postLanguage', 'postLocation', 'postSentiment',
                              'allPostTags', 'postPermalink', 'autoTagNumTags', 'autoTagTags',
                              'manualTagNumTags', 'manualTagTags', 'workQueueAtTimeOfEvent',
                              'uniqueId', 'authorAutoTags', 'authorManualTags', 'authorUUID'])
        print("The file you are try to convert does not exist in the folder ./downloads")

if __name__ == '__main__':
import argparse

class OptionsConvertFile:
    Defines options relevant for interacting with the file converter.

    def __init__(self):
        def parse_format_arg(format):
            if len(format.split(".")) > 1 and  format.split(".")[1].casefold() == "json":
                return format
                raise argparse.ArgumentTypeError("Not a valid format: {}, try to load a file with '.json' extension.".format(format))

        parser = argparse.ArgumentParser(description="A basic example to convert JSON files to CSV and skipping some columns")

        path_opts = parser.add_argument_group('Path file', 'Where to find the file to convert.')
        path_opts.add_argument('-f', '--file', required=True, help='File name to convert.', type=parse_format_arg)

        parsed = parser.parse_args()

        self.fileName = parsed.file

Did this page help you?