Skip to content

Instantly share code, notes, and snippets.

@AndrewWCarson
Created August 2, 2019 00:29
Show Gist options
  • Save AndrewWCarson/a99ac7324745e89cd2aa45dec719a1b6 to your computer and use it in GitHub Desktop.
Save AndrewWCarson/a99ac7324745e89cd2aa45dec719a1b6 to your computer and use it in GitHub Desktop.
An example iterating through events in the Addigy API and outputting them to a .csv file.
#!/usr/local/bin/python3
# Addigy API Credentials
client_id = 'YOUR ADDIGY API CLIENT_ID HERE'
client_secret = 'YOUR ADDIGY API CLIENT_SECRET HERE'
# Libraries
import sys
import requests
import json
import csv
import os
def get_events(payload):
response = requests.get('https://prod.addigy.com/api/system/events', params=payload)
if response.status_code != HTTPS_SUCCESS:
sys.exit(REQUEST_ERR + str(response.status_code))
return json.loads(response.text)
def fix_unicode(data):
if isinstance(data, unicode):
return data.encode('utf-8')
elif isinstance(data, dict):
data = dict((fix_unicode(k), fix_unicode(data[k])) for k in data)
elif isinstance(data, list):
for i in xrange(0, len(data)):
data[i] = fix_unicode(data[i])
return data
# Error & Success Codes for API stuff
HTTPS_SUCCESS=200
REQUEST_ERR = 'API Error: '
# Variables
outputPath = '/tmp/addigy_events.csv'
outputMode = 'w'
allEvents = list()
events = list()
if len(sys.argv) > 1:
if len(sys.argv) > 2:
sys.exit('Error: too many arguments.\nOnly one argument will be accepted for the output file path.\nTry: python export_events.py /path/to/file.csv')
else:
outputPath = sys.argv[1]
if os.path.isfile(outputPath):
outputMode = 'a'
# Continue making calls until there are no more events
while True:
if len(allEvents) == 0:
payload = {'client_id': client_id, 'client_secret': client_secret}
else:
payload = {'client_id': client_id, 'client_secret': client_secret, 'object_id': lastID}
events = get_events(payload)
if len(events) == 0:
break
allEvents += events
lastID = events[len(events)-1]['_id']
# Get rid of that Unicode stuff
allEvents = fix_unicode(allEvents)
with open(outputPath, outputMode) as outputFile:
writer = csv.writer(outputFile)
if outputMode == 'w':
writer.writerow(['Date', 'Sender', 'Action', 'Receiver', 'Result', 'ID', 'Level', 'OrgID'])
for event in allEvents:
writer.writerow([event['date'], event['action_sender'], event['action'], event['action_receiver'], event['result'], event['_id'], event['level'], event['orgid']])
@AndrewWCarson
Copy link
Author

The new /api/system/events endpoint can be useful for a number of things, the primary use-case being maintaining a historical record of Addigy-related activity. Since the Addigy Events DB has a TTL on all records (90 days iirc), exporting the events in this manner can be useful for keeping records.

@AndrewWCarson
Copy link
Author

This will write to /tmp/addigy_events.csv unless you pass a file path as an argument. If the argument path already exists, it will append the query to the existing file.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment