Mixpanel Async Python Library

Common errors

• TypeError: __init__() got an unexpected keyword argument 'extra_arg'

  cause Attempting to pass arbitrary `*args` or `**kwargs` to the `AsyncBufferedConsumer` constructor, which is no longer supported in v0.3.0+.
  fix Remove any non-standard or arbitrary arguments from your `AsyncBufferedConsumer` initialization. Only use the explicitly defined parameters listed in the documentation.

• TypeError: flush() got an unexpected keyword argument 'async'

  cause Using the old `async` keyword argument with the `flush` method after version 0.2.0.
  fix Rename the `async` argument to `async_` when calling `flush()` (e.g., `await mixpanel.flush(async_=True)`).

• Events are not appearing in Mixpanel, or data is incomplete.

  cause The `AsyncBufferedConsumer` buffers events, and they may not have been sent before the application closed.
  fix Ensure that `await mixpanel.flush()` and `await mixpanel.close()` are called before your application fully exits. Wrap your event tracking logic in a `try...finally` block to guarantee these calls.

Warnings

• breaking The `AsyncBufferedConsumer` constructor no longer accepts arbitrary `*args` or `**kwargs`. All parameters must be explicitly defined arguments.
  Fix: Review your `AsyncBufferedConsumer` initialization. Remove any `*args` or `**kwargs` that were previously passed and ensure only named, supported parameters are used. New parameters like `import_url`, `request_timeout`, etc., are now directly supported.
• breaking The `flush` argument for the `AsyncBufferedConsumer.flush` method was renamed from `async` to `async_` to avoid conflicts with Python 3.7's `async` keyword.
  Fix: If you were explicitly passing `async=True` to `flush`, change it to `async_=True`.
• gotcha Events tracked with `MixpanelAsync` are buffered by default. If the application exits before the buffer is full or the flush interval passes, events might be lost.
  Fix: Always call `await mixpanel.flush()` to send any remaining events in the buffer, and `await mixpanel.close()` to clean up resources, before your application terminates. This should typically be done in a `finally` block or a shutdown hook.

Install

• pip install mixpanel-py-async Install stable version

Imports

• MixpanelAsync

  from mixpanel_async import MixpanelAsync

• AsyncBufferedConsumer

  from mixpanel_async import AsyncBufferedConsumer

Quickstart

This example demonstrates how to initialize `MixpanelAsync`, track an event, update user profile properties, and increment a user property. It highlights the importance of calling `flush()` and `close()` to ensure all buffered events are sent and resources are properly cleaned up before application exit.

import asyncio
import os
from mixpanel_async import MixpanelAsync

async def main():
    # Retrieve Mixpanel project token from environment variables
    # For a real application, ensure MIXPANEL_PROJECT_TOKEN is set
    project_token = os.environ.get("MIXPANEL_PROJECT_TOKEN", "YOUR_MIXPANEL_PROJECT_TOKEN")

    if project_token == "YOUR_MIXPANEL_PROJECT_TOKEN":
        print("Warning: MIXPANEL_PROJECT_TOKEN environment variable not set. Using a placeholder.")

    # Initialize Mixpanel with the project token. AsyncBufferedConsumer is used by default.
    mixpanel = MixpanelAsync(project_token)

    try:
        user_id = "test_user_123"
        # Track an event
        await mixpanel.track(user_id, "Product Viewed", {"product_id": "SKU456", "category": "Electronics"})

        # Update user profile properties
        await mixpanel.people_set(user_id, {"$first_name": "Async", "$last_name": "User", "Plan": "Free"})

        # Increment a user property
        await mixpanel.people_increment(user_id, {"Views": 1})

        print(f"Tracking data for user {user_id}...")

    finally:
        # Crucial: flush any remaining buffered events before the application exits
        await mixpanel.flush()
        # Close the consumer's HTTP session and clean up resources
        await mixpanel.close()
        print("Mixpanel events flushed and consumer closed.")

if __name__ == "__main__":
    asyncio.run(main())