Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Mixpanel Delete People
This is from mixpanel customer service, I just PEP8ified it. Update api key, secret and token.
You can define which users you want to delete on line 125. Right now it will delete users who haven't
been seen in over 7 weeks. You will recieve a confirmation prompt before the people are deleted.
import hashlib
import time
import urllib
import base64
import sys
import json
except ImportError:
import simplejson as json
import eventlet
from import urllib2
except ImportError:
print "You need to pip install eventlet. Quitting now..."
class Mixpanel(object):
def __init__(self, api_key, api_secret, token):
self.api_key = api_key
self.api_secret = api_secret
self.token = token
def request(self, params, format='json'):
'''let's craft the http request'''
params['api_key'] = self.api_key
params['expire'] = int(time.time())+600 # 600 is ten minutes from now
if 'sig' in params:
del params['sig']
params['sig'] = self.hash_args(params)
request_url = '' + self.unicode_urlencode(params)
request = urllib.urlopen(request_url)
data =
return data
def hash_args(self, args, secret=None):
'''Hash dem arguments in the proper way
join keys - values and append a secret -> md5 it'''
for a in args:
if isinstance(args[a], list):
args[a] = json.dumps(args[a])
args_joined = ''
for a in sorted(args.keys()):
if isinstance(a, unicode):
args_joined += a.encode('utf-8')
args_joined += str(a)
args_joined += "="
if isinstance(args[a], unicode):
args_joined += args[a].encode('utf-8')
args_joined += str(args[a])
hash = hashlib.md5(args_joined)
if secret:
elif self.api_secret:
return hash.hexdigest()
def unicode_urlencode(self, params):
''' Convert stuff to json format and correctly handle unicode url parameters'''
if isinstance(params, dict):
params = params.items()
for i, param in enumerate(params):
if isinstance(param[1], list):
params[i] = (param[0], json.dumps(param[1]),)
result = urllib.urlencode([(k, isinstance(v, unicode) and v.encode('utf-8') or v) for k, v in params])
return result
def update(self, userlist, uparams):
url = ""
batch = []
for user in userlist:
distinctid = user['$distinct_id']
tempparams = {
'token': self.token,
'$distinct_id': distinctid,
'$ignore_alias': True
payload = {"data": base64.b64encode(json.dumps(batch)), "verbose": 1, "api_key": self.api_key}
response = urllib2.urlopen(url, urllib.urlencode(payload))
message =
'''if something goes wrong, this will say what'''
if json.loads(message)['status'] != 1:
print message
def batch_update(self, users, params):
pool = eventlet.GreenPool(size=200)
while len(users):
batch = users[:50]
pool.spawn(self.update, batch, params)
users = users[50:]
print "Done!"
if __name__ == '__main__':
api = Mixpanel(
parameters = {}
'''Here is the place to define your selector to target only the users that you're after'''
'selector': '(datetime(%s - %s) > properties["$last_seen"])' % (int(round(time.time() * 1000)), (1000 * 60 * 60 * 24 * 7 * 7))
response = api.request(parameters)
'session_id': json.loads(response)['session_id'],
'page': 0
global_total = json.loads(response)['total']
print "Here are the # of people %d" % global_total
fname = "Mixpanel_backup-" + str(int(time.time())) + ".txt"
has_results = True
total = 0
print fname
f = open(fname, 'w')
x = raw_input("You are about to delete %s users. Enter YES to continue: " % (str(global_total)))
while has_results and x == "YES":
responser = json.loads(response)['results']
total += len(responser)
has_results = len(responser) == 1000
for data in responser:
print "%d / %d" % (total, global_total)
api.batch_update(responser, {'$delete': ''})
parameters['page'] += 1
if has_results:
response = api.request(parameters)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.