mixpanel.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import requests
import os
import json


READ_API = 'https://mixpanel.com/api/2.0/engage'
WRITE_API = 'https://api.mixpanel.com/engage'
secret = user = os.environ.get('MIXPANEL_USER', 'foo')
token = password = os.environ.get('MIXPANEL_PASSWORD', 'bar')


def modify_user(distinct_id):
    """
    Adds some dummy attributes to a user for the purposes of deletion
    """
    payload = {
        '$token': f'{token}',
        '$distinct_id': f'{distinct_id}',
        '$set': {
            '$first_name': 'Big',
            '$last_name': 'Boss',
            '$email': '[email protected]',
            'Address': 'Outer Haven'
        }
    }

    r = requests.post(
        url=WRITE_API,
        headers={'Content-type': 'application/json'},
        auth=(user, password),
        data=json.dumps(payload))

    r.raise_for_status()

    if r.text == '0':
        print(f'There was a problem updating [{distinct_id}]. That\'s all we know.')


def delete_user(distinct_id):
    """
    Delete a single user from Mixpanel

    https://developer.mixpanel.com/docs/http#storing-user-profiles
    https://developer.mixpanel.com/docs/http#delete
    """
    payload = {
        '$token': f'{token}',
        '$distinct_id': f'{distinct_id}',
        '$delete': ''}

    r = requests.post(
        url=WRITE_API,
        auth=(user, password),
        data=json.dumps(payload))

    r.raise_for_status()

    if r.text == '0':
        print(f'There was a problem deleting [{distinct_id}]. That\'s all we know.')


def get_users():
    """
    Get all the users (even without profiles) from Mixpanel
    https://developer.mixpanel.com/docs/data-export-api#engage
    """
    users = []
    params = {
        'include_all_users': 'true',
        'filter_by_cohort': '{"raw_cohort":{"behaviors":{},"description":"","name":"","selector":{"children":[{"children":[{"property":"predefined_cohort","value":"$all_users"},{"property":"literal","value":false}],"operator":"or"},{"property":"literal","value":true}],"operator":"and"},"is_visible":true,"is_locked":false}}',
        'limit': 1000,
        'page': 0,
        'session_id': None}
    print('Getting users...')
    while True:
        r = requests.get(
            READ_API,
            auth=(user, password),
            params=params)

        data = r.json()

        for r in data['results']:
            if not r['$properties']:
                users.append(r)
            else:
                continue

        if 'session_id' not in data or len(data['results']) == 0:
            break
        else:
            params['session_id'] = data['session_id']
            params['page'] += 1

    return users


if __name__ == '__main__':
    users = get_users()
    print(f'Found {len(users)} users')
    for u in users:
        modify_user(u['$distinct_id'])
        delete_user(u['$distinct_id'])