Initial lib

.gitignore

@@ -0,0 +1 @@
+*.pyc

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)

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()