diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0d20b64 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +*.pyc diff --git a/mixpanel.py b/mixpanel.py new file mode 100644 index 0000000..7d94f46 --- /dev/null +++ b/mixpanel.py @@ -0,0 +1,104 @@ +import base64 +import json +import urllib +import urllib2 + +class Mixpanel(object): + + def __init__(self, token, base_url='https://api.mixpanel.com/'): + """ + To use mixpanel, create a new Mixpanel object using your token. + Use this object to start tracking. + Example: + mp = Mixpanel('36ada5b10da39a1347559321baf13063') + """ + self._token = token + self._base_url = base_url + + @classmethod + def _prepare_data(self, data): + return urllib.urlencode({'data': base64.b64encode(json.dumps(data)),'verbose':1}) + + def _write_request(self, base_url, endpoint, request, batch=False): + """ + Writes a request taking in either 'track/' for events or 'engage/' for + people. + """ + data = self._prepare_data(request) + request_url = ''.join([base_url, endpoint]) + try: + if not batch: + response = urllib2.urlopen(request_url, data).read() + else: + batch_request = urllib2.Request(request_url, data) + response = urllib2.urlopen(batch_request).read() + + except urllib2.HTTPError as e: + raise e + return response == '1' + + def _people(self, distinct_id, update_type, properties): + record = { + '$token': self._token, + '$distinct_id': distinct_id, + update_type: properties, + } + return self._write_request(self._base_url, 'engage/', record) + + def track(self, event_name, properties={}): + """ + For basic event tracking. Should pass in name of event name and + dictionary of properties. + Example: + mp.track('clicked button', { 'color': 'blue', 'text': 'no' }) + """ + all_properties = { 'token' : self._token } + all_properties.update(properties) + event = { + 'event': event_name, + 'properties': all_properties, + } + return self._write_request(self._base_url, 'track/', event) + + def alias(self, alias_id, original): + """ + Allows you to set a custom alias for people records. + Example: + mp.alias('amy@mixpanel.com', '13793') + """ + record = { + 'event': '$create_alias', + 'properties': { + 'distinct_id': original, + 'alias': alias_id, + 'token': self._token, + } + } + return self._write_request(self._base_url, 'engage/', record) + + def people_set(self, distinct_id, properties): + return self._people(distinct_id, '$set', properties) + + def people_set_once(self, distinct_id, properties): + return self._people(distinct_id, '$set_once', properties) + + def people_add(self, distinct_id, properties): + return self._people(distinct_id, '$add', properties) + + def people_append(self, distinct_id, properties): + return self._people(distinct_id, '$append', properties) + + def people_union(self, distinct_id, properties): + return self._people(distinct_id, '$union', properties) + + def people_unset(self, distinct_id, properties): + return self._people(distinct_id, '$unset', properties) + + def people_delete(self, distinct_id): + return self._people(distinct_id, '$delete', "") + + def send_people_batch(self, data): + return self._write_request(self._base_url, 'engage/', data, batch=True) + + def send_events_batch(self, data): + return self._write_request(self._base_url, 'track/', data, batch=True) diff --git a/test.py b/test.py new file mode 100755 index 0000000..bc2bcbf --- /dev/null +++ b/test.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python +import urllib +import unittest +from mixpanel import Mixpanel +from mock import Mock, patch + +class MixpanelTestCase(unittest.TestCase): + track_request_url = 'https://api.mixpanel.com/track/' + engage_request_url = 'https://api.mixpanel.com/engage/' + + def test_constructor(self): + token = '12345' + mp = Mixpanel(token) + self.assertEqual(mp._token, token) + + def test_prepare_data(self): + prepared_ab = Mixpanel._prepare_data({'a': 'b'}) + self.assertEqual('data=eyJhIjogImIifQ%3D%3D&verbose=1', prepared_ab) + + def test_track(self): + token = '12345' + mp = Mixpanel(token) + mock_response = Mock() + mock_response.read.return_value = '1' + with patch('urllib2.urlopen', return_value = mock_response) as mock_urlopen: + mp.track('button press', {'size': 'big', 'color': 'blue'}) + data = mp._prepare_data({'event': 'button press', 'properties': {'token': '12345', 'size': 'big', 'color': 'blue'}}) + mock_urlopen.assert_called_once_with(self.track_request_url, data) + + def test_people_set(self): + token = '12345' + mp = Mixpanel(token) + mock_response = Mock() + mock_response.read.return_value = '1' + with patch('urllib2.urlopen', return_value = mock_response) as mock_urlopen: + mp.people_set('amq', {'birth month': 'october', 'favorite color': 'purple'}) + data = mp._prepare_data({'$token': '12345', '$distinct_id': 'amq', '$set': {'birth month': 'october', 'favorite color': 'purple'}}) + mock_urlopen.assert_called_once_with(self.engage_request_url, data) + + def test_alias(self): + token = '12345' + mp = Mixpanel(token) + mock_response = Mock() + mock_response.read.return_value = '1' + with patch('urllib2.urlopen', return_value = mock_response) as mock_urlopen: + mp.alias('amq','3680') + data = mp._prepare_data({'event': '$create_alias', 'properties': {'distinct_id': '3680', 'alias': 'amq', 'token': '12345'}}) + mock_urlopen.assert_called_once_with(self.engage_request_url, data) + + def test_events_batch(self): + events_list = [ + { + "event": "Signed Up", + "properties": { + "distinct_id": "13793", + "token": "e3bc4100330c35722740fb8c6f5abddc", + "Referred By": "Friend", + "time": 1371002000 + } + }, + { + "event": "Uploaded Photo", + "properties": { + "distinct_id": "13793", + "token": "e3bc4100330c35722740fb8c6f5abddc", + "Topic": "Vacation", + "time": 1371002104 + } + } + ] + token = "e3bc4100330c35722740fb8c6f5abddc" + mp = Mixpanel(token) + mock_response = Mock() + mock_response.read.return_value = '1' + data = mp._prepare_data(events_list) + with patch('urllib2.Request', return_value = mock_response) as mock_Request: + with patch('urllib2.urlopen', return_value = mock_response) as mock_urlopen: + mp.send_events_batch(events_list) + mock_Request.assert_called_once_with(self.track_request_url, data) + +if __name__ == "__main__": + unittest.main()