Convenience functions for accessing Mixpanel API

gistfile1.py

import json

DELAY = 300
MAX_ATTEMPTS = 2

MIX_API_KEY = settings['mixpanel_key']
MIX_API_SECRET = settings['mixpanel_secret']

def hash_args(args, secret=None):
    import hashlib
    for a in args:
        if isinstance(args[a], list): args[a] = json.dumps(args[a])

    args_joined = ''
    for a in sorted(args.keys()):
        if isinstance(a, unicode):
            args_joined += a.encode('utf-8')
        else:
            args_joined += str(a)

        args_joined += '='

        if isinstance(args[a], unicode):
            args_joined += args[a].encode('utf-8')
        else:
            args_joined += str(args[a])

    hash = hashlib.md5(args_joined)

    if secret:
        hash.update(secret)
    elif self.api_secret:
        hash.update(self.api_secret)
    return hash.hexdigest()

def mixpanel_try(endpoint, args, attempts=0):
    import requests
    from time import sleep
    try:
        events = requests.post("http://mixpanel.com/api/2.0/{}".format(endpoint), args, timeout=DELAY)
        return json.loads(events.content)
    except:
        if attempts < MAX_ATTEMPTS:
            sleep(DELAY)
            return mixpanel_try(endpoint, args, attempts+1)
        else:
            raise

def mixpanel_get(endpoint, args, data='data'):
    import time
    args['api_key'] = MIX_API_KEY
    args['expire'] = int(time.time() + 33554432)
    args['sig'] = hash_args(args, MIX_API_SECRET)
    result = mixpanel_try(endpoint, args)
    if data in result:
        return result[data]
    else:
        return result