/
mixpanel.py
73 lines (54 loc) · 2.3 KB
/
mixpanel.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
#! /usr/bin/env python
#
# Mixpanel, Inc. -- http://mixpanel.com/
#
# Python API client library to consume mixpanel.com analytics data.
import md5
import hashlib
import urllib
import time
try:
import json
except ImportError:
import simplejson as json
class Mixpanel(object):
ENDPOINT = 'http://mixpanel.com/api'
VERSION = '1.0'
def __init__(self, api_key, api_secret):
self.api_key = api_key
self.api_secret = api_secret
def request(self, endpoint, method, params, format='json'):
"""
endpoint - Parent level access point (e.g. events, properties, etc.)
method - Child level access point (e.g. general, unique, etc.)
params - Extra parameters associated with method
"""
if not len(params): return False
params['api_key'] = self.api_key
params['expire'] = int(time.time()) + 600 # Grant this request 10 minutes.
params['format'] = format
params['sig'] = self.hash_args(params)
request_url = self.ENDPOINT + '/' + endpoint + '/' + str(self.VERSION) + '/' + method + '?' + self.unicode_urlencode(params)
request = urllib.urlopen(request_url)
return json.loads(request.read())
def unicode_urlencode(self, params):
if isinstance(params, dict):
params = params.items()
for i, k in enumerate(params):
if isinstance(k[1], list):
params[i] = (k[0], json.dumps(k[1]),)
return urllib.urlencode([(k, isinstance(v, unicode) and v.encode('utf-8') or v) for k, v in params])
def hash_args(self, args, secret=None):
"""
Hashes arguments by joining key=value pairs, appending a secret, and then taking the MD5 hex digest.
"""
for a in args:
if isinstance(args[a], list): args[a] = json.dumps(args[a])
args_joined = ''.join(['%s=%s' % (isinstance(x, unicode) and x.encode("utf-8") or x, isinstance(args[x], unicode) and
args[x].encode("utf-8") or args[x]) for x in sorted(args.keys())])
hash = hashlib.md5(args_joined)
if secret:
hash.update(secret)
elif self.api_secret:
hash.update(self.api_secret)
return hash.hexdigest()