Using MixPanel with Python

In my Recent post about my Great search for a decent analytics solution, I introduced MixPanel, the new Analytics system that we're test-driving at Newstex. I went over a little bit of how we've been testing it out by importing all the data we had in our old system into MixPanel so we could review it with our actual live data. While doing so, I decided to write something a little more generic and robust for tracking events on the server side. After all, it might be useful in other places within our system, so we can actually track events from the backend.

Tracking events server-side has a pretty big advantage in that it doesn't depend on a specific client application. Since we do everything as APIs with clients, this means that all of our clients log similar events no matter what platform they're running on.

We do everything in Python, and the documentation does give a relatively rudimentary API to pushing events to the server, but interacting directly to the REST API just seemed to be a lot easier. I wrote a very simple class that handles pushing events to the server, both asynchronously and synchronously. Instead of pushing off events through a system like is done mixpanel-celery this just spawns off a new thread for each event tracking if you call it with track_async. It also allows you to pass in a callback function to be executed once the track event is fired, which helps if you need to be absolutely sure your event was saved properly.

But enough talking, get to the code!:

"""
Event tracking, currently uses Mixpanel: https://mixpanel.com
"""
TRACK_BASE_URL = "http://api.mixpanel.com/track/?data=%s"
ARCHIVE_BASE_URL = "http://api.mixpanel.com/import/?data=%s&api_key=%s"
import urllib2
import json
import base64
import time

class EventTracker(object):
  """Simple Event Tracker
  Designed to be generic, but currently uses Mixpanel
  to actually handle the tracking of the events
  """
  def __init__(self, token, api_key=None):
    """Create a new event tracker
    :param token: The auth token to use to validate each request
    :type token: str
    """
    self.token = token
    self.api_key = api_key

  def track(self, event, properties=None, callback=None):
    """Track a single event
    :param event: The name of the event to track
    :type event: str
    :param properties: An optional dict of properties to describe the event
    :type properties: dict
    :param callback: An optional callback to execute when
      the event has been tracked.
      The callback function should accept two arguments, the event
      and properties, just as they are provided to this function 
      This is mostly used for handling Async operations
    :type callback: function
    """
    if properties is None:
      properties = {}
    if not properties.has_key("token"):
      properties['token'] = self.token
    if not properties.has_key("time"):
      properties['time'] = int(time.time())

    assert(properties.has_key("distinct_id")), "Must specify a distinct ID"

    params = {"event": event, "properties": properties}
    data = base64.b64encode(json.dumps(params))
    if self.api_key:
      resp = urllib2.urlopen(ARCHIVE_BASE_URL % (data, self.api_key))
    else:
      resp = urllib2.urlopen(TRACK_BASE_URL % data)
    resp.read()

    if callback is not None:
      callback(event, properties)

  def track_async(self, event, properties=None, callback=None):
    """Track an event asyncrhonously, essentially this runs the track
    event in a new thread
    :param event: The name of the event to track
    :type event: str
    :param properties: An optional dict of properties to describe the event
    :type properties: dict
    :param callback: An optional callback to execute when the event has been
      tracked. The callback function should accept two arguments, the event
      and properties, just as they are provided to this function
    :type callback: function

    :return: Thread object that will process this request
    :rtype: :class:`threading.Thread`
    """
    from threading import Thread
    t = Thread(target=self.track, kwargs={
      'event': event, 
      'properties': properties, 
      'callback': callback
    })
    t.start()
    return t

Usage is incredibly simple:

tracker = EventTracker(TOKEN)
tracker.track_async("My Event", {
   "distinct_id": "some_unique_id", 
   "mp_tag_name": "My User Name",
   "my_property": "some value"
   "some_int_value": 0,
})

It's important to note that you should convert datetimes into integers as unix timestamps as MixPanel handles those very well.

See https://mixpanel.com/docs/properties-or-segments/special-or-reserved-properties for special properties that can be used.