-
-
Save trentniemeyer/cdf976a64630111ca90741944965a24c to your computer and use it in GitHub Desktop.
#! /usr/bin/env python | |
# | |
# Mixpanel, Inc. -- http://mixpanel.com/ | |
# | |
# Python API client library to consume mixpanel.com analytics data. | |
# | |
# Copyright 2010-2013 Mixpanel, Inc | |
# | |
# Licensed under the Apache License, Version 2.0 (the "License"); | |
# you may not use this file except in compliance with the License. | |
# You may obtain a copy of the License at | |
# | |
# http://www.apache.org/licenses/LICENSE-2.0 | |
# | |
# Unless required by applicable law or agreed to in writing, software | |
# distributed under the License is distributed on an "AS IS" BASIS, | |
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
# See the License for the specific language governing permissions and | |
# limitations under the License. | |
import base64 | |
import urllib.request | |
try: | |
import json | |
except ImportError: | |
import simplejson as json | |
class Mixpanel(object): | |
ENDPOINT = 'https://data.mixpanel.com/api' | |
VERSION = '2.0' | |
def __init__(self, api_secret): | |
self.api_secret = api_secret | |
def request(self, methods, params, http_method='GET', format='json'): | |
""" | |
methods - List of methods to be joined, e.g. ['events', 'properties', 'values'] | |
will give us http://mixpanel.com/api/2.0/events/properties/values/ | |
params - Extra parameters associated with method | |
""" | |
request_url = '/'.join([self.ENDPOINT, str(self.VERSION)] + methods) | |
if http_method == 'GET': | |
data = None | |
request_url = request_url + '/?' + self.unicode_urlencode(params) | |
else: | |
data = self.unicode_urlencode(params) | |
auth = base64.b64encode(self.api_secret).decode("ascii") | |
headers = {'Authorization': 'Basic {encoded_secret}'.format(encoded_secret=auth)} | |
request = urllib.request.Request(request_url, data, headers) | |
response = urllib.request.urlopen(request, timeout=120) | |
str_response = response.read().decode('utf8') | |
lines = str_response.splitlines(True) | |
records = [] | |
for line in lines: | |
obj = json.loads(line) | |
records.append(obj) | |
return records | |
def unicode_urlencode(self, params): | |
""" | |
Convert lists to JSON encoded strings, and correctly handle any | |
unicode URL parameters. | |
""" | |
if isinstance(params, dict): | |
params = list(params.items()) | |
for i,param in enumerate(params): | |
if isinstance(param[1], list): | |
params.remove(param) | |
params.append ((param[0], json.dumps(param[1]),)) | |
return urllib.parse.urlencode( | |
[(k, v) for k, v in params] | |
) | |
if __name__ == '__main__': | |
encoded_secret = b'REPLACE_ME' | |
api = Mixpanel(api_secret=encoded_secret) | |
data = api.request(['export'], { | |
'event': ['Product-Detail-View'], | |
'to_date': "2016-06-16", | |
'from_date': "2016-05-31" | |
}) | |
print (json.dumps(data, indent=4)) |
update: i encoded api_secret in line 51 and managed to get pass the error. however i am now getting urllib.error.HTTPError: HTTP Error 502: Bad Gateway
update: I failed to notice that there are 2 separate end points. Change ENDPOINT = 'https://data.mixpanel.com/api' to 'https://mixpanel.com/api', depending on which endpoint you will be utilizing
I got it working without any errors
import base64
import urllib.request
import ssl
try:
import json
except ImportError:
import simplejson as json
class Mixpanel(object):
ENDPOINT = 'https://mixpanel.com/api'
VERSION = '2.0'
def __init__(self, api_secret):
self.api_secret = api_secret
def request(self, methods, params, http_method='GET', format='json'):
"""
methods - List of methods to be joined, e.g. ['events', 'properties', 'values']
will give us http://mixpanel.com/api/2.0/events/properties/values/
params - Extra parameters associated with method
"""
params['format'] = format
# print(base64.b64encode(self.api_secret).decode("ascii"))
request_url = '/'.join([self.ENDPOINT, str(self.VERSION)] + methods)
if http_method == 'GET':
data = None
request_url = request_url + '/?' + self.unicode_urlencode(params)
else:
data = self.unicode_urlencode(params)
auth = base64.b64encode(self.api_secret).decode("ascii")
headers = {'Authorization': 'Basic {encoded_secret}'.format(encoded_secret=auth)}
request = urllib.request.Request(request_url, data, headers)
# print(request)
context = ssl._create_unverified_context()
response = urllib.request.urlopen(request, context=context, timeout=120)
str_response = response.read().decode('utf8')
lines = str_response.splitlines(True)
records = []
for line in lines:
obj = json.loads(line)
records.append(obj)
return records
def unicode_urlencode(self, params):
"""
Convert lists to JSON encoded strings, and correctly handle any
unicode URL parameters.
"""
if isinstance(params, dict):
params = list(params.items())
for i,param in enumerate(params):
if isinstance(param[1], list):
params.remove(param)
params.append ((param[0], json.dumps(param[1]),))
return urllib.parse.urlencode(
[(k, v) for k, v in params]
)
if __name__ == '__main__':
encoded_secret = b'SECRET_KEY'
# byteAPISecret = bytes(encoded_secret + ':', "utf-8")
api = Mixpanel(api_secret=encoded_secret)
data = api.request(['events'], {
'event': ['YOUR_EVENT'],
'unit': 'hour',
'interval': 24,
'type': 'general'
})
# data = api.request(['export'], {
# 'event': ['YOUR_EVENT'],
# 'to_date': "2018-04-16",
# 'from_date': "2018-05-31"
# })
print (json.dumps(data, indent=4))
@Harshakvarma Great, thank you!
I am getting this error <HTTPError 500: 'Internal Server Error'> on the line
response = urllib.request.urlopen(request, context=context, timeout=120)
please give suggetion on this
thanks
I got an error 'urllib.error.HTTPError: HTTP Error 402: Payment Required' This means we have to buy the pro version?
Hi, in my case, the error I have is that when entering the "Event" parameter, I got "event array is not valid json". I tried different ways, but unable to get it.
could someone help me here?
hey im getting the error TypeError: a bytes-like object is required, not 'str'. Does anyone know why? Seems like the error is because of line 51. Should I encode that line instead?