mixpanel-py-async

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())