How to migrate your PostHog self-hosted to PostHog Cloud

#!/usr/bin/env python
import copy
import time
import uuid
from urllib.parse import urlparse

import requests

# Your source PostHog instance
source_posthog_scheme_and_host = 'https://posthog.your-domain.tld'

# Generate a personal API key to read the data from your source PostHog instance
source_api_key = 'xxx'

# Your project API key provided by PostHog in your project settings
dest_project_api_key = 'xxx'


def is_valid(key: str, data: dict) -> bool:
    """
    Helper function to check if the value is empty or none
    :param key: data key to check
    :param value: dict
    :return: true if valid (not empty, None), false otherwise
    """
    if key not in data or not data[key] or str(data[key]).strip().lower() == 'none':
        return False

    return True


def clean_source_data(results: [dict]) -> [dict]:
    """
    Clean up data from PostHog source
    Note: this function do not mutate `results`
    :param results:
    :return:
    """
    _results = []
    if not results:
        return _results

    distinct_id = 'distinct_id'

    for result in results:
        data = copy.deepcopy(result)
        del data['id']

        if not is_valid(distinct_id, data):
            data[distinct_id] = str(uuid.uuid4())

        if 'properties' in data:
            properties = data['properties']
            if not is_valid(distinct_id, properties):
                # copy distinct_id value from the parent object
                properties[distinct_id] = data[distinct_id]

        _results.append(data)

    return _results


def capture(data: [dict], count: int = 0):
    """
    Function to send data to the dest PostHog instance
    :param data:
    :param count:
    :return:
    """
    res = requests.post('https://app.posthog.com/capture', json={'api_key': dest_project_api_key, 'batch': data},
                        headers={'Content-type': 'application/json'})

    if res.status_code != 200:
        if count >= 100:
            print('retry exceeded')
            exit(1)
        time.sleep(3)
        print('Retry sending data (status code: {}) to dest PostHog with data {}'.format(res.status_code, data))
        return capture(data, count + 1)

    return res


def get_source_data() -> dict:
    headers = {'Authorization': 'Bearer {}'.format(source_api_key)}

    url = '{}/api/event'.format(source_posthog_scheme_and_host)

    with open('posthog_completed_urls', 'a') as f:
        while 1:
            query = urlparse(url).query
            if query:
                # PostHog return weird next URL with tons of 'before' params
                url = '{}/api/event?before={}'.format(source_posthog_scheme_and_host, query.split('=')[-1])

            res = requests.get(url, headers=headers)
            if res.status_code == 200:
                j_res = res.json()
                _data = clean_source_data(j_res['results'])

                yield _data

                f.write(url + '\n')
                url = j_res['next']
            else:
                print('Retry fetching events (status code: {}) from PostHog source with URL {}'.format(res.status_code, url))
                time.sleep(3)


if __name__ == '__main__':
    for data in get_source_data():
        capture(data)
        print('{} lines migrated'.format(len(data)))

    print('ok')