EventFlows

A lightweight, robust, and privacy-first server-side tracking library for Python.

EventFlows acts as a universal router for your conversion events. You define your tracking event once, and EventFlows automatically cleans it, hashes sensitive PII, batches it in a background thread, and safely dispatches it to multiple external APIs (Meta, GA4, Reddit, TikTok) without slowing down your main application.

Key Features

• Zero Latency: Events are dropped into a thread-safe memory queue instantly. Your users never wait for a tracking pixel to load.
• Privacy by Design: Emails and phone numbers are automatically normalized and SHA-256 hashed before they ever leave your server.
• Smart Payload Mapping: Send custom data freely. EventFlows dynamically formats it for "Open" APIs (like Meta) and strictly filters it for "Rigid" APIs (like Reddit).
• Destination Filtering: Choose exactly which networks receive which events.
• Resilient Networking: Built-in exponential backoff and retry logic for network blips or rate limits.

Installation

Currently, EventFlows is installed from source.

Using Poetry

poetry add git+https://github.com/YOUR_USERNAME/EventFlows.git

Using pip

pip install git+https://github.com/YOUR_USERNAME/EventFlows.git

Quickstart

Register your providers once at the start of your application, then use tracker.track anywhere.

import logging
from EventFlows import tracker, EventType
from EventFlows import ConsoleProvider, MetaProvider, RedditProvider, GA4Provider

# Setup logging to see EventFlows processing in the background
logging.basicConfig(level=logging.INFO)

# 1. Register your destinations (Providers)
tracker.add_provider(ConsoleProvider())  # Great for local debugging
tracker.add_provider(MetaProvider(pixel_id="123", access_token="abc"))
tracker.add_provider(GA4Provider(measurement_id="G-123", api_secret="abc"))

# 2. Track an event anywhere in your app
# This happens instantly. Hashing and API calls occur in a background thread.
tracker.track(
    name="ProSubscription",
    event_type=EventType.PURCHASE,
    user_data={
        "email": "  User@Example.com  ",  # Automatically trimmed, lowercased, and SHA-256 hashed
        "ip_address": "192.168.1.1"
    },
    properties={
        "value": 49.99,
        "currency": "USD",
        "plan_type": "annual"
    },
    event_id="ORD-9999"  # Used for server-side deduplication
)

# 3. Graceful Shutdown
# Essential for ensuring the memory queue is flushed before the process exits.
# Call this in your app's shutdown/cleanup hook.
tracker.shutdown()

Why use tracker.shutdown()?

EventFlows uses an internal queue to batch events and keep your application fast. If your script or server exits abruptly without calling shutdown(), events still in the queue may not be delivered.

• FastAPI/Starlette: Use the @app.on_event("shutdown") decorator (or lifespan context manager).
• Flask: Use atexit.register(tracker.shutdown).
• Scripts: Call it as the final line of your execution.