Metadata-Version: 2.4
Name: analytics-tracker
Version: 0.1.0
Summary: Async analytics tracking library with PostgreSQL/TimescaleDB support
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: asyncpg>=0.29.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: testcontainers>=4.14.0; extra == "dev"

# 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

```bash
pip install analytics-tracker
```

## Quick Start

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

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

### SendResult

Return type for track operations:

```python
@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:

```python
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:

```python
@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:

```python
@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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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
