Integrate with Mixpanel API in Python

Mixpanel is one of the leading business analytics services. But expensive.

ref:
https://mixpanel.com/report/1262196/dashboard

Installation

$ pip install mixpanel==4.3.2

ref:
https://pypi.org/project/mixpanel/

Configuration

You probably don't want to send Mixpanel events synchronously and blocking your application, so just put them into a Celery task. Maybe with a shared reqeusts session.

# in app.py
from celery import Celery
from flask import Flask
import boltons.cacheutils
import requests

def make_celery(app):
    celery = Celery(
        'YOUR_APP_NAME',
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL'],
    )

    class ContextTask(celery.Task):
        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                # pass an extra `task` parameter in each Celery task for accessing Task properties
                return celery.Task.__call__(self, *args, **kwargs)

        @boltons.cacheutils.cachedproperty
        def session(self):
            session = requests.Session()
            session.mount('https://', requests.adapters.HTTPAdapter(
                pool_maxsize=10,
                pool_connections=10,
            ))
            return session

        @boltons.cacheutils.cachedproperty
        def mixpanel(self):
            import mixpanel

            class RequestsConsumer(mixpanel.Consumer):

                def __init__(self, client=None, *args, **kwargs):
                    self.client = client or requests
                    super(RequestsConsumer, self).__init__(*args, **kwargs)

                def _write_request(self, request_url, json_message, api_key=None):
                    data = {
                        'data': mixpanel.base64.b64encode(json_message.encode('utf8')),
                        'verbose': 1,
                        'ip': 0,
                    }
                    if api_key:
                        data.update({'api_key': api_key})

                    try:
                        response = self.client.post(
                            request_url, data=data, timeout=self._request_timeout,
                        )
                        response.raise_for_status()
                    except requests.exceptions.HTTPError as err:
                        raise mixpanel.MixpanelException(err) from err

                    try:
                        data = response.json()
                    except ValueError:
                        raise mixpanel.MixpanelException(f'Cannot interpret Mixpanel server response: {response}')
                    else:
                        if data['status'] != 1:
                            raise mixpanel.MixpanelException(f"Mixpanel error: {data['error']}")
                    return True

            mp = mixpanel.Mixpanel('MIXPANEL_PROJECT_TOKEN', consumer=RequestsConsumer(client=self.session))
            return mp

    celery.Task = ContextTask
    return celery

app = Flask('simple-project')
app.celery = make_celery(app)

ref:
https://github.com/mixpanel/mixpanel-python
http://docs.celeryproject.org/en/latest/reference/celery.app.task.html

Usage

ref:
https://mixpanel.com/help/reference/python
https://mixpanel.github.io/mixpanel-python/

Property Data Types

Event and user properties support following data types:

  • String
  • Number
  • Boolean
  • Date
  • List

Mixpanel does not allow dict in properties.

ref:
https://help.mixpanel.com/hc/en-us/articles/115004547063-Properties-Supported-Data-Types

Attach Extra Attributes To User Profiles

# ext_mixpanel/tasks.py
import datetime

import celery
import mixpanel

@celery.shared_task(bind=True)
def people_set(task, *args, **kwargs):
    return task.mixpanel.people_set(*args, **kwargs)

people_set.apply_async(kwargs=dict(
    distinct_id=str(user.id),
    properties={
        '$email': user.email,
        'username': user.username,
        'signed_up_at': mixpanel.serialize(datetime.datetime.utcnow()),
    },
))

ref:
https://mixpanel.com/help/reference/python#creating-profiles
https://mixpanel.github.io/mixpanel-python/#mixpanel.Mixpanel.people_set

Track Events

# ext_mixpanel/tasks.py
import celery

@celery.shared_task(bind=True)
def track(task, *args, **kwargs):
    return task.mixpanel.track(*args, **kwargs)

track.apply_async(kwargs=dict(
    distinct_id=post.author.id,
    event_name='post.sent',
    properties={
        'post.id': post.id,
        'post.title': post.title,
        'author.id': post.author.id,
        'author.username': post.author.username,
        'tags': [str(tag) for tag in post.tags],
    },
))

ref:
https://mixpanel.com/help/reference/python#sending-events
https://mixpanel.github.io/mixpanel-python/#mixpanel.Mixpanel.track

Track Revenues

# ext_mixpanel/tasks.py
import celery

@celery.shared_task(bind=True)
def people_track_charge(task, *args, **kwargs):
    return task.mixpanel.people_track_charge(*args, **kwargs)

people_track_charge.apply_async(kwargs=dict(
    distinct_id=order.customer.id,
    amount=round(order.amount_in_usd, 2),
    properties={
        'payment_method': order.payment_method,
    },
))

ref:
https://mixpanel.com/help/reference/python#tracking-revenue
https://mixpanel.github.io/mixpanel-python/#mixpanel.Mixpanel.people_track_charge

JQL

function main() {
  return Events(
    {
      event_selectors: [{event: 'user.registered', selector: ''}],
      from_date: '2018-09-22',
      to_date: '2018-09-22'
    },
  )
  .reduce(mixpanel.reducer.count());
}

ref:
https://mixpanel.com/help/reference/jql/api-reference