Dead simple event tracking with celery and git

base.py

"""
    Simple mixpanel wrapper. Does everything synchronously as we will be
    calling this from a worker queue
"""

import base64
import json
import urllib
from django.conf import settings

API_TOKEN = getattr(settings, "MIXPANEL_API_TOKEN")
API_URL = getattr(settings, "MIXPANEL_API_URL")

def track(event, properties=None):
    """Track mixpanel events"""

    if properties is None:
        properties = {}

    properties["token"] = API_TOKEN

    if "time" in properties:
        properties["time"] = int(properties["time"])

    params = {"event": event, "properties": properties}

    data = base64.b64encode(json.dumps(params))
    conn = urllib.urlopen(API_URL + data)

    contents = conn.read()

    assert contents == "1", "Mixpanel returned unknown value: %s" % contents

    return True

demo.py

from mixpanel.tasks import TrackEvent

# Inside a view
def my_view(request):
    TrackEvent.delay("search_view", {
        "ip": request.META['REMOTE_ADDR'],
        "distinct_id": session_id,
        "time": time.time()})

tasks.py

"""
    Mixpanel tasks
"""

from celery.task import Task, tasks
from base import track


class TrackEvent(Task):
    """Asynchrnously track a Mixpanel event"""

    def run(self, event, properties=None, *args, **kwargs):
        return track(event, properties)

tasks.register(TrackEvent)