Created
August 2, 2019 00:29
-
-
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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']]) |
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
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.