mixpanel_posthog.py

from mixpanel import Mixpanel, Consumer, MixpanelException

class PostHogConsumer(Consumer):
    _posthog_endpoints = {
        'events': 'https://t.posthog.com/capture',
        'people': 'https://t.posthog.com/engage',
        'groups': 'https://t.posthog.com/groups',
        'imports': 'https://t.posthog.com/import'
    }
    _posthog_api_key = '<posthog_key>'

    def send(self, endpoint, json_message=None, api_key=None):
        if endpoint in self._endpoints:
            self._write_request(self._posthog_endpoints[endpoint], json_message, self._posthog_api_key)
            self._write_request(self._endpoints[endpoint], json_message, api_key)
        else:
            raise MixpanelException('No such endpoint "{0}". Valid endpoints are one of {1}'.format(endpoint, self._endpoints.keys()))

mp = Mixpanel('<mixpanel_key>', consumer=PostHogConsumer())

mp.people_set('distinct_id', {'email': '[email protected]'})
mp.track('distinct_id', 'some event')