analytics-tracker 0.1.0

Async analytics tracking library with PostgreSQL/TimescaleDB support

Analytics Tracker

A robust, async analytics tracking library for Python applications with PostgreSQL and TimescaleDB support.

Features

• Event Tracking: Track user events with rich properties and automatic timestamps
• Async Batch Processing: Efficient background worker for database persistence
• Thread-Safe Buffer: Event buffer with configurable size and flush intervals
• Plugin System: Transform events with custom plugins
• Connection Pooling: Efficient PostgreSQL/TimescaleDB connection management
• Graceful Shutdown: Automatic flush on shutdown with error aggregation
• Retry Logic: Automatic retry for temporary database connection errors
• Metrics Collection: Track performance and event statistics

Installation

pip install analytics-tracker

Quick Start

import asyncio
from analytics_tracker import EventTracker, DBConfig, BufferConfig

async def main():
    db_config = DBConfig(
        host="localhost",
        port=5432,
        database="analytics",
        user="postgres",
        password="secret",
        min_size=1,
        max_size=10,
    )
    buffer_config = BufferConfig(
        max_size=1000,
        flush_interval=5.0,
        batch_size=100,
    )

    tracker = EventTracker(
        buffer_config=buffer_config,
        table_name="business_events",
    )

    await tracker.initialize(db_config)
    await tracker.run_migrations()
    await tracker.start()

    tracker.track("user123", "page_view", {"page": "/home"})
    tracker.track("user123", "button_click", {"button": "submit"})

    with tracker.user("user456") as user:
        user.track("purchase", {"amount": 99.99, "currency": "USD"})

    await tracker.close()

asyncio.run(main())

Configuration

DBConfig

Database connection configuration with validation:

from analytics_tracker import DBConfig

config = DBConfig(
    host="localhost",       # Database host
    port=5432,              # Port (1-65535)
    database="analytics",   # Database name
    user="postgres",        # Username
    password="secret",      # Password (hidden in repr)
    min_size=1,             # Min pool size (>=1)
    max_size=10,            # Max pool size (>= min_size)
)

config = DBConfig.from_env()

BufferConfig

Event buffer configuration:

from analytics_tracker import BufferConfig

config = BufferConfig(
    max_size=1000,      # Max events in buffer
    flush_interval=5.0, # Seconds between flushes
    batch_size=100,     # Events per batch insert
)

Plugins

Transform events using plugins. Plugins receive an event dict and return a transformed dict:

from analytics_tracker import EventTracker

def add_timestamp_plugin(event: dict) -> dict:
    from datetime import datetime, timezone
    event["processed_at"] = datetime.now(timezone.utc).isoformat()
    return event

def add_utm_params(event: dict) -> dict:
    if "properties" not in event:
        event["properties"] = {}
    event["properties"]["utm_source"] = "newsletter"
    return event

tracker = EventTracker()
tracker.add_plugin(add_timestamp_plugin)
tracker.add_plugin(add_utm_params)

Environment Variables

All configuration can be set via environment variables:

Variable	Description	Default
DB_HOST	Database hostname	localhost
DB_PORT	Database port	5432
DB_NAME	Database name	analytics
DB_USER	Database user	postgres
DB_PASSWORD	Database password	postgres
DB_MIN_SIZE	Min pool size	1
DB_MAX_SIZE	Max pool size	10

API Reference

EventTracker

Main class for event tracking:

class EventTracker:
    def __init__(
        self,
        buffer_config: BufferConfig | None = None,
        table_name: str = "business_events",
        db_config: DBConfig | None = None,
    )

    async def initialize(db_config: DBConfig | None = None) -> None
    async def run_migrations() -> None
    async def start() -> None
    async def stop() -> None
    async def close() -> None

    def track(user_id: str, event_type: str, properties: dict | None = None) -> SendResult
    def add_plugin(plugin: Plugin) -> None
    def get_metrics() -> dict

    @contextmanager
    def user(user_id: str) -> UserTracker

UserTracker

Context manager for tracking events for a specific user:

class UserTracker:
    def track(event_type: str, properties: dict | None = None) -> SendResult

SendResult

Return type for track operations:

@dataclass
class SendResult:
    success: bool           # Whether event was queued
    event_id: UUID | None   # Event ID if successful
    error: str | None       # Error message if failed

EventModel

Event data model:

class EventModel:
    event_id: UUID          # Unique event identifier
    user_id: str            # User identifier
    event_type: str         # Type of event
    properties: dict        # Event properties
    timestamp: datetime     # Event timestamp (UTC)

BufferConfig

Configuration for event buffer:

@dataclass
class BufferConfig:
    max_size: int           # Max events in buffer (default: 1000)
    flush_interval: float   # Seconds between flushes (default: 5.0)
    batch_size: int         # Events per batch insert (default: 100)

DBConfig

Database connection configuration with validation:

@dataclass
class DBConfig:
    host: str               # Database host
    port: int               # Port (1-65535)
    database: str           # Database name
    user: str               # Username
    password: str           # Password
    min_size: int           # Min pool size (>=1)
    max_size: int           # Max pool size (>= min_size)

    @property
    def connection_url(self) -> str

MetricsCollector

Track performance metrics:

class MetricsCollector:
    def record_event_queued() -> None
    def record_dropped() -> None
    def record_batch_sent(batch_size: int, duration_ms: float) -> None
    def get_metrics() -> dict

MigrationManager

Handle database migrations:

class MigrationManager:
    table_name: str

    async def initialize(db_config: DBConfig) -> None
    async def run_migrations() -> None
    async def close() -> None

BackgroundWorker

Background worker for batch processing:

class BackgroundWorker:
    def __init__(
        self,
        buffer: EventBuffer,
        dsn: str,
        batch_size: int,
        flush_interval: float,
        metrics: MetricsCollector,
        connection_manager: DBConnectionManager,
        event_saver: EventSaver,
    )

    async def start() -> None
    async def stop() -> None
    async def async_flush() -> None

Metrics

Track performance metrics:

metrics = tracker.get_metrics()
# {
#     "events_queued_total": 500,
#     "events_dropped": 0,
#     "events_sent_total": 400,
#     "batches_sent_total": 4,
#     "avg_flush_time_ms": 45.2,
#     "buffer_size": 100,
#     "migrations_run": True,
# }

Exceptions

Custom exceptions for error handling:

from analytics_tracker import (
    TrackerError,      # Base exception
    ValidationError,   # Configuration validation errors
    BufferError,       # Buffer operation errors
    PluginError,       # Plugin execution errors
)

try:
    tracker.track("user123", "event", {})
except ValidationError as e:
    print(f"Invalid configuration: {e}")
except BufferError as e:
    print(f"Buffer error: {e}")

Aggregate error for graceful shutdown:

from analytics_tracker.tracker import AggregateError

try:
    await tracker.close()
except AggregateError as e:
    print(f"Multiple errors occurred: {e.errors}")

Architecture

User Code
    |
    v
EventTracker.track() --> EventBuffer --> BackgroundWorker --> EventSaver --> PostgreSQL
    |                           |                                          |
    +--> PluginManager ---------+                                          |
    |                                                                          |
    +--> MetricsCollector ----------------------------------------------+

License

MIT