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.
Prerequisites:
- 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 'main.py' file executable:
chmod u+x main.py && chmod u+x convertFromJsonToCsv.py
Upon executing the main.py 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 convertFromJsonToCsv.py 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 ./convertFromJsonToCsv.py -h.
#!/usr/local/bin/python
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:
client.download_report_to_tmp_file(download_url)
if __name__ == '__main__':
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),
auth=self.opts.auth)
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
else:
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 = requests.post(request_url, 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
else:
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'][
'detail']))
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'][
'runnerState']))
return True, report_status_response.json()['result']['jobInfo']['downloadUrl']
else:
print("report status is {} ... waiting".format(report_status_response.json()['result']['result']['runnerState']))
time.sleep(2)
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()))
else:
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)
os.close(fd)
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 "https://analytics-api.app.lithium.com"
elif region.casefold() == "eu-west-1":
return "https://analytics-api-emea.app.lithium.com"
else:
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 = parsed.company
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))
#!/usr/local/bin/python
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'])
else:
print("The file you are try to convert does not exist in the folder ./downloads")
if __name__ == '__main__':
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
else:
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
Updated over 4 years ago