Analytics Tracker

https://github.com/yoxan/analytics-tracker
Admin
A robust, async analytics tracking library for Python applications with PostgreSQL and TimescaleDB...
Tokens:21,375
Snippets:163
Trust Score:5.2
Update:8 months ago
Show doc for...
Tokens:
Raw
### Quick Start Event Tracking with Python

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Demonstrates the basic usage of the EventTracker for asynchronous event tracking. It shows initialization with database and buffer configurations, starting the tracker, tracking events, and graceful shutdown.

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

--------------------------------

### Async Python Test Example

Source: https://github.com/yoxan/analytics-tracker/blob/master/AGENTS.md

An example of an asynchronous test function using pytest-asyncio. This demonstrates how to initialize the EventTracker and assert its state after initialization, mocking necessary dependencies.

```python
@pytest.mark.asyncio
async def test_tracker_initialize():
    tracker = EventTracker(dsn="postgresql://...")
    await tracker.initialize(db_config)
    assert tracker.connection_manager is not None

```

--------------------------------

### Setup Database Schema with MigrationManager in Python

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

Illustrates how to use the MigrationManager to set up the TimescaleDB hypertable schema, including compression and indexing. This requires the analytics-tracker library and asyncio. The function `run_migrations` is idempotent.

```python
import asyncio
from analytics_tracker import EventTracker, DBConfig

async def setup_database():
    db_config = DBConfig(
        host="localhost",
        database="analytics",
        user="postgres",
        password="postgres",
    )

    # Migrations create:
    # - TimescaleDB extension
    # - Hypertable partitioned by timestamp
    # - Compression policy (7 days)
    # - Indexes on (user_id, timestamp) and (event_type, timestamp)

    tracker = EventTracker(db_config=db_config, table_name="custom_events")
    await tracker.initialize()
    await tracker.run_migrations()  # Safe to run multiple times (IF NOT EXISTS)

    print("Database schema ready")
    await tracker.close()

asyncio.run(setup_database())
```

--------------------------------

### Start Background Worker (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the asynchronous starting of the background worker for the EventTracker. It mocks the BackgroundWorker class and its start method, ensuring the worker is instantiated and started correctly.

```python
@pytest.mark.asyncio
async def test_tracker_start_stop(db_config):
    tracker = EventTracker(db_config=db_config)
    with patch.object(tracker, '_worker', None):
        with patch('analytics_tracker.tracker.BackgroundWorker') as MockWorker:
            mock_worker = MockMock()
            MockWorker.return_value = mock_worker
            await tracker.start()
            MockWorker.assert_called_once()
            mock_worker.start.assert_called_once()
```

--------------------------------

### Python Docstring Example

Source: https://github.com/yoxan/analytics-tracker/blob/master/AGENTS.md

Example of a Google-style docstring for a Python method, including arguments and return values. This adheres to the project's code style requirements for public methods.

```python
def track(self, user_id: str, event_type: str, properties: dict | None = None) -> SendResult:
    """Track an event for a user.

    Args:
        user_id: Unique identifier for the user.
        event_type: Type of the event (e.g., "page_view").
        properties: Optional event properties as key-value pairs.

    Returns:
        SendResult indicating success or failure with optional error message.
    """

```

--------------------------------

### Start and Stop BackgroundWorker (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the start and stop methods of the BackgroundWorker. It verifies that calling start() sets the is_running flag to True and creates a task, and that stop() sets is_running back to False.

```python
def test_worker_start(worker):
    worker.start()
    assert worker.is_running is True
    assert worker._task is not None

def test_worker_stop(worker):
    worker.start()
    worker.stop()
    assert worker.is_running is False
```

--------------------------------

### Initialize and Use Event Tracker (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-design.md

Demonstrates the API usage for initializing the EventTracker with database and buffer configurations, running migrations, starting the tracker, tracking events, retrieving metrics, and closing the tracker.

```python
from analytics_tracker import (
    EventTracker, DBConfig,
    BufferConfig
)

db_config = DBConfig(
    dsn="postgresql://user:pass@localhost:5432/mydb",
    pool_size=10,
    max_overflow=20,
    command_timeout=30.0
)

buffer_config = BufferConfig(
    max_size=5000,
    batch_size=200,
    flush_interval=10.0
)

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

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

result = tracker.track("user_123", "button_click", {"button": "submit"})

metrics = tracker.get_metrics()
print(metrics)

await tracker.close()
```

--------------------------------

### EventBuffer Usage Example in Python

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

Provides a starting point for using the EventBuffer class, which offers a thread-safe queue for events. This is an internal component of EventTracker but can be accessed directly for advanced scenarios. It requires the analytics-tracker library.

```python
from analytics_tracker.buffer import EventBuffer
from analytics_tracker.schema import EventModel
from uuid import uuid4
from datetime import datetime, timezone

# Example usage would go here, demonstrating adding events and retrieving them.
```

--------------------------------

### Install Analytics Tracker

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Installs the analytics-tracker library using pip. This is the primary method for adding the library to your Python project.

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

--------------------------------

### Test EventTracker Start/Stop with Mock BackgroundWorker

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the asynchronous start and stop functionality of the EventTracker using a mocked BackgroundWorker. It verifies that the worker's start method is called once during the tracker's start process.

```python
@pytest.mark.asyncio
async def test_tracker_start_stop(db_config, buffer_config, mock_background_worker):
    tracker = EventTracker(db_config=db_config, buffer_config=buffer_config)
    with patch('analytics_tracker.tracker.BackgroundWorker', return_value=mock_background_worker):
        await tracker.start()
        mock_background_worker.start.assert_called_once()
```

--------------------------------

### Test Full Event Tracking Pipeline (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the complete event tracking pipeline, including initialization, migration, starting the tracker, tracking events, and verifying metrics. It requires a running PostgreSQL instance and uses 'pytest-asyncio'.

```python
@pytest.mark.asyncio
async def test_full_pipeline(tracker):
    await tracker.initialize()
    
    try:
        await tracker.run_migrations()
        
        await tracker.start()
        
        result = tracker.track("user_1", "test_event", {"key": "value"})
        assert result.success is True
        
        result2 = tracker.track("user_2", "another_event", {"data": 123})
        assert result2.success is True
        
        await asyncio.sleep(2)  # Wait for flush
        
        metrics = tracker.get_metrics()
        assert metrics["events_queued_total"] >= 2
        
    finally:
        await tracker.close()
```

--------------------------------

### Initialize and Run BackgroundWorker

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Initializes the BackgroundWorker with event buffer, saver, and metrics. Starts the worker by creating an asyncio task for the run loop. The worker continuously flushes events at a specified interval.

```python
import asyncio
import time
from typing import Optional

from analytics_tracker.buffer import EventBuffer
from analytics_tracker.saver import EventSaver
from analytics_tracker.schema import EventModel


class BackgroundWorker:
    def __init__(
        self, 
        buffer: EventBuffer,
        event_saver: EventSaver,
        metrics: "MetricsCollector",
        batch_size: int = 100,
        flush_interval: float = 5.0,
    ):
        self.buffer = buffer
        self.event_saver = event_saver
        self.metrics = metrics
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.is_running = False
        self._task: Optional[asyncio.Task] = None

    def start(self) -> None:
        self.is_running = True
        self._task = asyncio.create_task(self._run_loop())

    def stop(self) -> None:
        self.is_running = False
        if self._task:
            self._task.cancel()
            try:
                asyncio.run(asyncio.wait_for(self._task, timeout=10.0))
            except asyncio.CancelledError:
                pass

    async def _run_loop(self) -> None:
        while self.is_running:
            await self._flush()
            await asyncio.sleep(self.flush_interval)
```

--------------------------------

### Get Metrics (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the get_metrics method to ensure it returns a dictionary containing all tracked metrics, including calculated average flush time.

```python
def test_metrics_get_metrics():
    m = MetricsCollector()
    m.record_event_queued()
    m.record_batch_sent(10, 5.0)
    metrics = m.get_metrics()
    assert metrics["events_queued_total"] == 1
    assert metrics["events_sent_total"] == 10
    assert metrics["events_dropped"] == 0
    assert metrics["events_dropped_batch"] == 0
    assert metrics["avg_flush_time_ms"] == 5.0
```

--------------------------------

### Implement Plugin System for Event Transformation (Python)

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

This code illustrates the plugin system for transforming events before they are queued. Plugins are functions that modify event dictionaries and are executed in the order they are added. Example plugins include adding processing metadata, enriching with UTM parameters, and sanitizing Personally Identifiable Information (PII).

```python
import asyncio
from datetime import datetime, timezone
from analytics_tracker import EventTracker, DBConfig

# Plugin to add processing metadata
def add_metadata_plugin(event: dict) -> dict:
    event["processed_at"] = datetime.now(timezone.utc).isoformat()
    event["sdk_version"] = "1.0.0"
    return event

# Plugin to enrich with UTM parameters
def add_utm_plugin(event: dict) -> dict:
    if "properties" not in event:
        event["properties"] = {}
    event["properties"]["utm_source"] = "newsletter"
    event["properties"]["utm_medium"] = "email"
    return event

# Plugin to sanitize PII
def sanitize_pii_plugin(event: dict) -> dict:
    props = event.get("properties", {})
    if "email" in props:
        props["email"] = "***@***.***"
    if "phone" in props:
        props["phone"] = "***-***-****"
    return event

async def main():
    db_config = DBConfig(host="localhost", database="analytics")
    tracker = EventTracker(db_config=db_config)

    # Add plugins (executed in order)
    tracker.add_plugin(add_metadata_plugin)
    tracker.add_plugin(add_utm_plugin)
    tracker.add_plugin(sanitize_pii_plugin)

    await tracker.initialize()
    await tracker.start()

    # Events will be transformed by all plugins
    tracker.track("user123", "signup", {"email": "user@example.com", "plan": "pro"})

    await tracker.close()

asyncio.run(main())
```

--------------------------------

### Start Background Worker for Event Processing (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Starts the background worker responsible for processing events from the buffer. It configures the worker with batch size and flush interval, raising a `RuntimeError` if the worker is already running. The worker handles flushing events to the database.

```python
async def start(self) -> None:
        if self._worker is not None:
            raise RuntimeError("Worker already started")

        batch_size = self.buffer_config.batch_size if self.buffer_config else 100
        flush_interval = self.buffer_config.flush_interval if self.buffer_config else 5.0

        self._worker = BackgroundWorker(
            buffer=self.buffer,
            event_saver=self.event_saver,
            metrics=self.metrics,
            batch_size=batch_size,
            flush_interval=flush_interval,
        )
        self._worker.start()
```

--------------------------------

### Get Metrics with No Flushes (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the get_metrics method when no batch flushes have occurred, ensuring the average flush time defaults to 0.0.

```python
def test_metrics_get_metrics_no_flushes():
    m = MetricsCollector()
    metrics = m.get_metrics()
    assert metrics["avg_flush_time_ms"] == 0.0
```

--------------------------------

### Implement Custom Event Transformation Plugins in Python

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Provides examples of creating custom plugins to transform events before they are tracked. Plugins are Python functions that receive an event dictionary and return a modified one.

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

--------------------------------

### Configure Event Buffer Behavior (Python)

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

The BufferConfig class allows customization of event buffering. You can define maximum buffer size, automatic flush intervals, and batch sizes for database inserts. This example shows configurations for high-throughput and low-latency scenarios, as well as the default settings.

```python
from analytics_tracker import BufferConfig

# High-throughput configuration
high_throughput_config = BufferConfig(
    max_size=10000,      # Large buffer for high event volume
    flush_interval=1.0,  # Frequent flushes
    batch_size=500,      # Large batches for efficiency
)

# Low-latency configuration
low_latency_config = BufferConfig(
    max_size=100,        # Small buffer
    flush_interval=0.5,  # Very frequent flushes
    batch_size=50,       # Smaller batches
)

# Default configuration (max_size=1000, flush_interval=5.0, batch_size=100)
default_config = BufferConfig()
```

--------------------------------

### Track Analytics Events with EventTracker (Python)

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

Demonstrates the primary interface for tracking analytics events using the EventTracker class. It covers configuration of database and buffer settings, initialization, starting the tracker, tracking individual events, retrieving metrics, and graceful shutdown.

```python
import asyncio
from analytics_tracker import EventTracker, DBConfig, BufferConfig

async def main():
    # Configure database connection
    db_config = DBConfig(
        host="localhost",
        port=5432,
        database="analytics",
        user="postgres",
        password="secret",
        min_size=2,
        max_size=10,
    )

    # Configure event buffering
    buffer_config = BufferConfig(
        max_size=1000,       # Max events in buffer before dropping
        flush_interval=5.0,  # Seconds between flushes
        batch_size=100,      # Events per batch insert
    )

    # Create tracker instance
    tracker = EventTracker(
        db_config=db_config,
        buffer_config=buffer_config,
        table_name="business_events",
    )

    # Initialize connection pool and run migrations
    await tracker.initialize()
    await tracker.run_migrations()
    await tracker.start()

    # Track events (synchronous, returns immediately)
    result = tracker.track("user123", "page_view", {"page": "/home", "referrer": "google"})
    if result.success:
        print(f"Event queued: {result.event_id}")
    else:
        print(f"Failed to queue event: {result.error}")

    # Track more events
    tracker.track("user123", "button_click", {"button": "submit", "form": "signup"})
    tracker.track("user456", "purchase", {"amount": 99.99, "currency": "USD"})

    # Get metrics
    metrics = tracker.get_metrics()
    # {'buffer_size': 3, 'events_queued_total': 3, 'events_sent_total': 0,
    #  'events_dropped': 0, 'avg_flush_time_ms': 0.0, 'migrations_run': True}

    # Graceful shutdown (flushes pending events)
    await tracker.close()

asyncio.run(main())

```

--------------------------------

### Minimal Implementation for __init__.py Exports

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

The implementation of the __init__.py file in the analytics_tracker package. It imports necessary components from various submodules and defines the `__all__` list to control public exports, ensuring that classes like DBConfig are available at the package level.

```python
# src/analytics_tracker/__init__.py
from analytics_tracker.buffer import EventBuffer
from analytics_tracker.config import BufferConfig
from analytics_tracker.db_config import DBConfig
from analytics_tracker.migrations import MigrationManager
from analytics_tracker.plugins import PluginManager
from analytics_tracker.result import SendResult
from analytics_tracker.schema import EventModel
from analytics_tracker.tracker import EventTracker, UserTracker
from analytics_tracker.worker import BackgroundWorker

__all__ = [
    "EventTracker",
    "UserTracker",
    "BufferConfig",
    "DBConfig",
    "SendResult",
    "EventModel",
    "EventBuffer",
    "BackgroundWorker",
    "MigrationManager",
    "PluginManager",
    "MetricsCollector",
]
```

--------------------------------

### Configure PostgreSQL Integration Tests (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Sets up fixtures for database and buffer configurations, and an EventTracker instance for integration tests. It requires the 'asyncpg' library and 'pytest-asyncio' for asynchronous testing.

```python
import pytest
import asyncio
from analytics_tracker import EventTracker, DBConfig, BufferConfig


@pytest.fixture
def db_config():
    return DBConfig(
        dsn="postgresql://postgres:postgres@localhost:5432/testdb",
        pool_size=2,
        max_overflow=5,
        command_timeout=30.0
    )


@pytest.fixture
def buffer_config():
    return BufferConfig(
        max_size=100,
        flush_interval=1.0,
        batch_size=10
    )


@pytest.fixture
def tracker(db_config, buffer_config):
    return EventTracker(
        db_config=db_config,
        buffer_config=buffer_config,
        table_name="test_events"
    )
```

--------------------------------

### Configure Database Connection (DBConfig) in Python

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Shows how to configure database connection parameters using DBConfig. It includes manual instantiation and loading configuration from environment variables.

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

--------------------------------

### Initialize EventTracker with DB and Buffer Configurations (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the successful initialization of the EventTracker with both database and buffer configurations. It asserts that the provided configurations are correctly assigned to the tracker instance and that the default table name is set.

```python
def test_tracker_init_with_config(db_config, buffer_config):
    tracker = EventTracker(db_config=db_config, buffer_config=buffer_config)
    assert tracker.db_config == db_config
    assert tracker.buffer_config == buffer_config
    assert tracker.table_name == "business_events"
```

--------------------------------

### Initialize EventTracker with DB Configuration (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Initializes the `EventTracker` with necessary database configuration. It requires `db_config` and `db_config.dsn` to be provided, raising a `ValueError` if they are missing. It also sets up buffering, plugins, metrics, and database connection management.

```python
from analytics_tracker.config import BufferConfig
from analytics_tracker.connection import DBConnectionManager
from analytics_tracker.db_config import DBConfig
from analytics_tracker.metrics import MetricsCollector
from analytics_tracker.plugins import PluginManager
from analytics_tracker.saver import EventSaver
from analytics_tracker.schema import EventModel
from analytics_tracker.worker import BackgroundWorker
from analytics_tracker.buffer import EventBuffer
from analytics_tracker.migrations import MigrationManager


class EventTracker:
    def __init__(
        self,
        buffer_config: Optional[BufferConfig] = None,
        db_config: Optional[DBConfig] = None,
        table_name: str = "business_events",
    ):
        if db_config is None:
            raise ValueError("db_config is required")
        if db_config.dsn is None:
            raise ValueError("db_config.dsn is required")

        self.db_config = db_config
        self.buffer_config = buffer_config
        self.table_name = table_name

        max_size = buffer_config.max_size if buffer_config else 1000
        self.buffer = EventBuffer(max_size=max_size)
        self.plugins = PluginManager()
        self.metrics = MetricsCollector()

        self.connection_manager = DBConnectionManager(self.db_config)
        self.event_saver = EventSaver(self.connection_manager, self.table_name)
        self.migration_manager = MigrationManager(
            self.connection_manager, self.table_name
        )

        self._worker: Optional[BackgroundWorker] = None
```

--------------------------------

### Initialize and Run Migrations with MigrationManager (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the initialization of the MigrationManager and its ability to run database migrations. It mocks the database connection and verifies that the correct SQL commands for creating extensions, tables, and hypertables are executed. Dependencies include pytest, unittest.mock, and the analytics_tracker library.

```python
import pytest
from unittest.mock import Mock, AsyncMock
from analytics_tracker.migrations import MigrationManager
from analytics_tracker.connection import DBConnectionManager
from analytics_tracker.db_config import DBConfig

@pytest.fixture
def db_config():
    return DBConfig(dsn="postgresql://localhost:5432/testdb")

@pytest.fixture
def connection_manager(db_config):
    mgr = DBConnectionManager(db_config)
    mgr._pool = AsyncMock()
    return mgr

def test_migration_manager_init(connection_manager):
    mgr = MigrationManager(connection_manager, "custom_table")
    assert mgr.connection_manager == connection_manager
    assert mgr.table_name == "custom_table"

@pytest.mark.asyncio
async def test_migration_manager_run(connection_manager):
    mgr = MigrationManager(connection_manager, "business_events")
    mock_conn = AsyncMock()
    connection_manager._pool.acquire.return_value.__aenter__.return_value = mock_conn
    
    await mgr.run_migrations()
    
    mock_conn.execute.assert_called_once()
    sql = mock_conn.execute.call_args[0][0]
    assert "CREATE EXTENSION IF NOT EXISTS timescaledb" in sql
    assert "CREATE TABLE IF NOT EXISTS business_events" in sql
    assert "timescaledb.create_hypertable" in sql
    assert "business_events" in sql

def test_migration_manager_get_sql():
    from analytics_tracker.connection import DBConnectionManager
    from analytics_tracker.db_config import DBConfig
    
    config = DBConfig(dsn="postgresql://localhost:5432/db")
    mgr = MigrationManager(Mock(), "test_table")
    sql = mgr._get_create_table_sql()
    
    assert "test_table" in sql
    assert "event_id" in sql
    assert "user_id" in sql
    assert "event_type" in sql
    assert "properties" in sql
    assert "timestamp" in sql
    assert "created_at" in sql

```

--------------------------------

### Run Pytest for Dependency and Configuration Tests

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

These bash commands execute tests using the 'pytest' framework. The first command runs all tests in 'test_dependencies.py', while the second specifically targets the 'test_asyncpg_available' function to verify the 'asyncpg' dependency. The third command runs tests for 'test_db_config.py', and the final command runs a specific test within that file to verify the 'DBConfig' creation and immutability.

```bash
pytest tests/unit/test_dependencies.py -v
```

```bash
pytest tests/unit/test_dependencies.py::test_asyncpg_available -v
```

```bash
pytest tests/unit/test_db_config.py -v
```

```bash
pytest tests/unit/test_db_config.py::test_db_config_creation -v
```

--------------------------------

### Commit Changes to Git

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

This command sequence demonstrates how to stage and commit changes to a Git repository. It adds specified files to the staging area and then creates a commit with a descriptive message, following conventional commit message standards ('deps: add asyncpg dependency' and 'feat: add DBConfig for database settings').

```bash
git add pyproject.toml tests/unit/test_dependencies.py
git commit -m "deps: add asyncpg dependency"
```

```bash
git add src/analytics_tracker/db_config.py tests/unit/test_db_config.py
git commit -m "feat: add DBConfig for database settings"
```

--------------------------------

### Run Integration Tests (Bash)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Command to execute integration tests for the database. It can be used to verify functionality with or without a PostgreSQL instance available, resulting in SKIP/FAIL or PASS respectively.

```bash
pytest tests/integration/test_db_integration.py -v
```

--------------------------------

### Retrieve Performance Metrics (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Demonstrates how to get the current performance metrics from the MetricsCollector. The output is a dictionary containing various tracking statistics.

```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,
# }
```

--------------------------------

### Initialize and Save Events with EventSaver (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Demonstrates the initialization of the EventSaver class and the process of saving a list of EventModel objects to a specified database table. It utilizes asyncpg for efficient bulk data insertion.

```python
import pytest
from unittest.mock import Mock, AsyncMock
from uuid import uuid4
from datetime import datetime
from analytics_tracker.saver import EventSaver
from analytics_tracker.connection import DBConnectionManager
from analytics_tracker.db_config import DBConfig
from analytics_tracker.schema import EventModel

@pytest.fixture
def db_config():
    return DBConfig(dsn="postgresql://localhost:5432/testdb")

@pytest.fixture
def connection_manager(db_config):
    mgr = DBConnectionManager(db_config)
    mgr._pool = AsyncMock()
    return mgr

@pytest.fixture
def event_saver(connection_manager):
    return EventSaver(connection_manager, "business_events")

@pytest.fixture
def sample_events():
    return [
        EventModel(
            event_id=uuid4(),
            user_id="user_1",
            event_type="test_event",
            properties={"key": "value"},
            timestamp=datetime.utcnow()
        ),
        EventModel(
            event_id=uuid4(),
            user_id="user_2",
            event_type="test_event_2",
            properties={"key2": "value2"},
            timestamp=datetime.utcnow()
        )
    ]

@pytest.mark.asyncio
async def test_event_saver_init(connection_manager):
    saver = EventSaver(connection_manager, "test_table")
    assert saver.connection_manager == connection_manager
    assert saver.table_name == "test_table"

@pytest.mark.asyncio
async def test_event_saver_save_empty(event_saver):
    result = await event_saver.save_events([])
    assert result == 0

@pytest.mark.asyncio
async def test_event_saver_save_events(event_saver, sample_events):
    mock_conn = AsyncMock()
    event_saver.connection_manager._pool.acquire.return_value.__aenter__.return_value = mock_conn
    
    result = await event_saver.save_events(sample_events)
    
    assert result == 2
    mock_conn.copy_records_to_table.assert_called_once()
    call_args = mock_conn.copy_records_to_table.call_args
    assert call_args.kwargs['table_name'] == "business_events"
    assert len(call_args.kwargs['records']) == 2
    assert call_args.kwargs['columns'] == (
        "event_id", "user_id", "event_type",
        "properties", "timestamp", "created_at"
    )

```

```python
# src/analytics_tracker/saver.py
import json
from typing import List

import asyncpg

from analytics_tracker.connection import DBConnectionManager
from analytics_tracker.schema import EventModel


class EventSaver:
    def __init__(self, connection_manager: DBConnectionManager, table_name: str):
        self.connection_manager = connection_manager
        self.table_name = table_name

    async def save_events(self, events: List[EventModel]) -> int:
        if not events:
            return 0

        records = [
            (
                str(e.event_id),
                e.user_id,
                e.event_type,
                json.dumps(e.properties),
                e.timestamp,
                e.created_at
            )
            for e in events
        ]

        async with self.connection_manager.pool.acquire() as conn:
            await conn.copy_records_to_table(
                self.table_name,
                records=records,
                columns=(
                    "event_id", "user_id", "event_type",
                    "properties", "timestamp", "created_at"
                )
            )

        return len(events)

```

--------------------------------

### Get Tracker Metrics (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the retrieval of metrics from the EventTracker. It asserts that the get_metrics method returns a dictionary containing expected metric keys like buffer_size and events_queued_total.

```python
def test_tracker_get_metrics(db_config):
    tracker = EventTracker(db_config=db_config)
    tracker.track("user_123", "test_event")
    metrics = tracker.get_metrics()
    assert "buffer_size" in metrics
    assert "events_queued_total" in metrics
```

--------------------------------

### Mock BackgroundWorker Fixture for Pytest

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Defines a Pytest fixture to create a mock BackgroundWorker. It mocks the start method to be synchronous and the stop method to be asynchronous, suitable for testing asynchronous operations.

```python
@pytest.fixture
def mock_background_worker():
    worker = Mock()
    worker.start = Mock()
    worker.stop = AsyncMock()
    return worker
```

--------------------------------

### Implement MigrationManager (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Minimal implementation of the MigrationManager class. It includes an asynchronous method `run_migrations` to execute SQL statements for setting up TimescaleDB extensions, creating tables, and configuring hypertables. The `_get_create_table_sql` method generates the necessary SQL. Dependencies include DBConnectionManager.

```python
# src/analytics_tracker/migrations.py
from analytics_tracker.connection import DBConnectionManager


class MigrationManager:
    def __init__(
        self,
        connection_manager: DBConnectionManager,
        table_name: str = "business_events",
    ):
        self.connection_manager = connection_manager
        self.table_name = table_name

    async def run_migrations(self) -> None:
        sql = self._get_create_table_sql()
        async with self.connection_manager.pool.acquire() as conn:
            await conn.execute(sql)

    def _get_create_table_sql(self) -> str:
        return f"""
            CREATE EXTENSION IF NOT EXISTS timescaledb;

            CREATE TABLE IF NOT EXISTS {self.table_name} (
                id BIGSERIAL,
                event_id UUID NOT NULL,
                user_id VARCHAR(255) NOT NULL,
                event_type VARCHAR(255) NOT NULL,
                properties JSONB DEFAULT '{{}}',
                timestamp TIMESTAMPTZ NOT NULL,
                created_at TIMESTAMPTZ DEFAULT NOW(),
                PRIMARY KEY (event_id, timestamp)
            );

            SELECT timescaledb.create_hypertable('{self.table_name}', 'timestamp');

            CREATE INDEX IF NOT EXISTS idx_{self.table_name}_user_id_timestamp
                ON {self.table_name} (user_id, timestamp DESC);

            CREATE INDEX IF NOT EXISTS idx_{self.table_name}_event_type_timestamp
                ON {self.table_name} (event_type, timestamp DESC);
        """

```

--------------------------------

### Initialize EventTracker with Default Buffer Configuration (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the initialization of the EventTracker when only the database configuration is provided, verifying that the buffer configuration defaults to None and the background worker is initially not set.

```python
def test_tracker_init_defaults(db_config):
    tracker = EventTracker(db_config=db_config)
    assert tracker.buffer_config is None
    assert tracker.table_name == "business_events"
    assert tracker._worker is None
```

--------------------------------

### Configure Database Connection from Environment Variables (Python)

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

This snippet demonstrates how to configure database connection parameters by reading environment variables. It sets up host, database name, and password for the analytics tracker. Ensure these environment variables are set before running the code.

```python
import os
os.environ["DB_HOST"] = "localhost"
os.environ["DB_NAME"] = "analytics"
os.environ["DB_PASSWORD"] = "secret"
config_from_env = DBConfig.from_env()
```

--------------------------------

### Initialize Database Connection (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the asynchronous initialization of the database connection manager within the EventTracker. It uses a patch to mock the connection manager's initialize method and verifies it's called once.

```python
@pytest.mark.asyncio
async def test_tracker_initialize(db_config):
    tracker = EventTracker(db_config=db_config)
    with patch.object(tracker.connection_manager, 'initialize', new_callable=AsyncMock) as mock_init:
        await tracker.initialize()
        mock_init.assert_called_once()
```

--------------------------------

### Run All Project Tests (Bash)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Executes all tests within the project to ensure all functionalities are working as expected after linting and formatting. Expected output is all tests passing.

```bash
pytest -v
```

--------------------------------

### Add asyncpg Dependency using uv

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

This snippet demonstrates adding the 'asyncpg' library to project dependencies using the 'uv' package manager. It's a common step in Python projects requiring asynchronous PostgreSQL interaction. The process involves running a command-line instruction.

```bash
uv add asyncpg
```

--------------------------------

### Initialize MetricsCollector (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the initialization of the MetricsCollector class, verifying that all internal counters and buffer size are set to their default zero values.

```python
import pytest
import threading
import time
from analytics_tracker.metrics import MetricsCollector

def test_metrics_collector_init():
    m = MetricsCollector()
    assert m.events_queued_total == 0
    assert m.events_sent_total == 0
    assert m.events_dropped == 0
    assert m.events_dropped_batch == 0
    assert m._buffer_size == 0
```

--------------------------------

### Run Database Migrations (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tests the asynchronous execution of database migrations using the EventTracker. It mocks the migration manager's run_migrations method and asserts that it is called exactly once.

```python
@pytest.mark.asyncio
async def test_tracker_run_migrations(db_config):
    tracker = EventTracker(db_config=db_config)
    with patch.object(tracker.migration_manager, 'run_migrations', new_callable=AsyncMock) as mock_migrate:
        await tracker.run_migrations()
        mock_migrate.assert_called_once()
```

--------------------------------

### Configure Event Buffer (BufferConfig) in Python

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Illustrates the configuration options for the event buffer, including maximum size, flush interval, and batch size for database inserts.

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

--------------------------------

### Configure Database Connection with DBConfig (Python)

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

Shows how to create and use the DBConfig dataclass for setting up PostgreSQL/TimescaleDB connection pool parameters. It includes programmatic configuration and demonstrates accessing connection URLs with and without passwords.

```python
from analytics_tracker import DBConfig

# Create configuration programmatically
db_config = DBConfig(
    host="db.example.com",
    port=5432,
    database="analytics_prod",
    user="analytics_user",
    password="secure_password",
    min_size=5,   # Minimum connections in pool
    max_size=20,  # Maximum connections in pool
)

# Access connection URL (includes password)
print(db_config.connection_url)
# postgresql://analytics_user:secure_password@db.example.com:5432/analytics_prod

# Safe URL for logging (password masked)
print(db_config.connection_url_without_password)
# postgresql://analytics_user:***@db.example.com:5432/analytics_prod

# Load from environment variables

```

--------------------------------

### Run Tests for Migrations (Bash)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Command to execute the unit tests for the migration module using pytest. This command helps in verifying the correctness of the MigrationManager implementation and its interaction with the database.

```bash
pytest tests/unit/test_migrations.py -v
```

--------------------------------

### Bash Commands for Testing and Linting

Source: https://github.com/yoxan/analytics-tracker/blob/master/AGENTS.md

Common bash commands for running tests, measuring code coverage, performing type checking with mypy, and linting with ruff. These commands are essential for maintaining code quality and verifying functionality.

```bash
# Run tests
python -m pytest tests/unit/ -v

# Run with coverage
python -m pytest tests/unit/ --cov=src/analytics_tracker

# Type checking (if mypy available)
python -m mypy src/analytics_tracker/

# Lint (if ruff available)
ruff check src/analytics_tracker/

```

--------------------------------

### Test DBConnectionManager Initialization and Pool Management (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Unit tests for the DBConnectionManager class, verifying initialization, pool creation, closing, and access to the pool property. It uses pytest and unittest.mock for asynchronous testing and patching.

```python
import pytest
from unittest.mock import Mock, AsyncMock, patch
from analytics_tracker.connection import DBConnectionManager
from analytics_tracker.db_config import DBConfig

@pytest.fixture
def db_config():
    return DBConfig(dsn="postgresql://localhost:5432/testdb")

@pytest.mark.asyncio
async def test_connection_manager_init(db_config):
    mgr = DBConnectionManager(db_config)
    assert mgr.db_config == db_config
    assert mgr._pool is None

@pytest.mark.asyncio
async def test_connection_manager_initialize(db_config):
    mgr = DBConnectionManager(db_config)
    with patch('analytics_tracker.connection.asyncpg.create_pool') as mock_pool:
        mock_pool.return_value = AsyncMock()
        await mgr.initialize()
        mock_pool.assert_called_once_with(
            dsn="postgresql://localhost:5432/testdb",
            min_size=5,
            max_size=15,
            command_timeout=30.0
        )
        assert mgr._pool is not None

@pytest.mark.asyncio
async def test_connection_manager_close(db_config):
    mgr = DBConnectionManager(db_config)
    mock_pool = AsyncMock()
    mgr._pool = mock_pool
    await mgr.close()
    mock_pool.close.assert_called_once()

@pytest.mark.asyncio
async def test_connection_manager_close_no_pool(db_config):
    mgr = DBConnectionManager(db_config)
    await mgr.close()  # Should not raise

@pytest.mark.asyncio
async def test_connection_manager_pool_property_uninitialized(db_config):
    mgr = DBConnectionManager(db_config)
    with pytest.raises(RuntimeError, match="Pool not initialized"):
        _ = mgr.pool

@pytest.mark.asyncio
async def test_connection_manager_pool_property(db_config):
    mgr = DBConnectionManager(db_config)
    mock_pool = AsyncMock()
    mgr._pool = mock_pool
    assert mgr.pool is mock_pool

```

--------------------------------

### Create and Manage Event Buffer

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

Demonstrates the creation of an `EventBuffer` with a maximum size, adding events, checking its fullness, peeking at the oldest event, removing a single event, and retrieving all events which also clears the buffer. The `add` method returns `False` if the buffer is full.

```python
from uuid import uuid4
from datetime import datetime, timezone

# Assuming EventBuffer, EventModel are defined elsewhere
# from your_module import EventBuffer, EventModel

# Placeholder classes for demonstration
class EventModel:
    def __init__(self, event_id, user_id, event_type, properties, timestamp):
        self.event_id = event_id
        self.user_id = user_id
        self.event_type = event_type
        self.properties = properties
        self.timestamp = timestamp

class EventBuffer:
    def __init__(self, max_size=100):
        self._max_size = max_size
        self._buffer = []

    def add(self, event):
        if len(self._buffer) >= self._max_size:
            return False
        self._buffer.append(event)
        return True

    def size(self):
        return len(self._buffer)

    def is_full(self):
        return len(self._buffer) >= self._max_size

    def peek(self):
        return self._buffer[0] if self._buffer else None

    def remove(self):
        return self._buffer.pop(0) if self._buffer else None

    def get_all(self):
        all_events = list(self._buffer)
        self._buffer = []
        return all_events

# Create buffer with max size
buffer = EventBuffer(max_size=100)

# Add events (returns False if full)
event = EventModel(
    event_id=uuid4(),
    user_id="user123",
    event_type="click",
    properties={"button": "submit"},
    timestamp=datetime.now(timezone.utc),
)

success = buffer.add(event)
print(f"Added: {success}, Size: {buffer.size()}")

# Check buffer state
print(f"Is full: {buffer.is_full()}")

# Peek at oldest event without removing
oldest = buffer.peek()
print(f"Oldest event type: {oldest.event_type if oldest else 'None'}")

# Remove single event
removed = buffer.remove()
print(f"Removed: {removed.event_id if removed else 'None'}")

# Get all events (clears buffer)
for i in range(10):
    buffer.add(EventModel(
        event_id=uuid4(),
        user_id=f"user{i}",
        event_type="batch_event",
        properties={{}},
        timestamp=datetime.now(timezone.utc),
    ))

all_events = buffer.get_all()
print(f"Got {len(all_events)} events, buffer size: {buffer.size()}")
```

--------------------------------

### Git Add and Commit (Bash)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Stages the modified metrics and test files and commits them with a descriptive message indicating the implementation of thread-safety.

```bash
git add src/analytics_tracker/metrics.py tests/unit/test_metrics.py
git commit -m "feat: make MetricsCollector thread-safe with Lock"
```

--------------------------------

### Manage User Sessions with UserTracker Context Manager (Python)

Source: https://context7.com/yoxan/analytics-tracker/llms.txt

Illustrates how to use the UserTracker context manager for tracking multiple events associated with a single user. This simplifies session tracking by eliminating the need to repeatedly specify the user ID.

```python
import asyncio
from analytics_tracker import EventTracker, DBConfig

async def track_user_session():
    db_config = DBConfig(host="localhost", database="analytics")
    tracker = EventTracker(db_config=db_config)
    await tracker.initialize()
    await tracker.start()

    # Track multiple events for same user with context manager
    with tracker.user("user789") as user:
        user.track("session_start", {"device": "mobile", "os": "iOS"})
        user.track("page_view", {"page": "/dashboard"})
        user.track("feature_used", {"feature": "export", "format": "csv"})
        result = user.track("session_end", {"duration_seconds": 342})
        print(f"Session tracked: {result.success}")

    await tracker.close()

asyncio.run(track_user_session())

```

--------------------------------

### Format Code with Ruff (Bash)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Applies code formatting using the Ruff formatter to the source and test directories, ensuring consistent code style.

```bash
ruff format src/ tests/
```

--------------------------------

### Manage Database Migrations (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Handles database schema migrations, including initialization, running migrations, and closing the connection. It requires a DBConfig object for initialization.

```python
class MigrationManager:
    table_name: str

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

--------------------------------

### Track User Event with Plugins and Buffering (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Tracks a user event by creating an `EventModel`, applying plugins, and attempting to add it to the event buffer. It records metrics for queued or dropped events and returns a `SendResult`. Handles exceptions during event processing.

```python
from datetime import datetime
from uuid import uuid4
from analytics_tracker.result import SendResult


class EventTracker:
    # ... (previous code) ...

    def _track(
        self,
        user_id: str,
        event_type: str,
        properties: Optional[dict]
    ) -> SendResult:
        event = EventModel(
            event_id=uuid4(),
            user_id=user_id,
            event_type=event_type,
            properties=properties or {},
            timestamp=datetime.utcnow()
        )

        try:
            event_dict = event.model_dump()
            event_dict = self.plugins.execute_plugins(event_dict, {})

            if self.buffer.add(event):
                self.metrics.record_event_queued()
                return SendResult(success=True, event_id=event.event_id)
            else:
                self.metrics.record_dropped()
                return SendResult(success=False, error="Buffer full")
        except Exception as e:
            return SendResult(success=False, error=str(e))
```

--------------------------------

### Run Pytest to Verify Event Saver Implementation (Bash)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

This command executes the unit tests for the EventSaver module using pytest. It's used to verify the correctness of the implementation after writing the code and tests.

```bash
pytest tests/unit/test_saver.py -v
```

--------------------------------

### Run Pytest Command

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Command to execute unit tests for the connection module using pytest. This is used to verify the correctness of the implementation against the defined test cases.

```bash
pytest tests/unit/test_connection.py -v

```

--------------------------------

### Create User Context for Tracking (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Provides a context manager for tracking events associated with a specific user. The `user` method yields a `UserTracker` instance, allowing events to be tracked within a `with` statement, simplifying user-specific event logging.

```python
from contextlib import contextmanager


class UserTracker:
    def __init__(self, tracker: "EventTracker", user_id: str):
        self._tracker = tracker
        self._user_id = user_id

    def track(self, event_type: str, properties: Optional[dict] = None) -> SendResult:
        return self._tracker._track(self._user_id, event_type, properties)

    def __enter__(self):
        return self

    def __exit__(self, *args):
        pass


class EventTracker:
    # ... (previous code) ...

    @contextmanager
    def user(self, user_id: str):
        yield UserTracker(self, user_id)
```

--------------------------------

### Run Database Migrations (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/docs/plans/2026-01-26-db-integration-implementation.md

Executes database migrations to ensure the event table schema is up-to-date. This asynchronous method uses the `MigrationManager` to apply necessary changes before or during the tracker's operation.

```python
async def run_migrations(self) -> None:
        await self.migration_manager.run_migrations()
```

--------------------------------

### Configure Event Buffer (Python)

Source: https://github.com/yoxan/analytics-tracker/blob/master/README.md

Sets up configuration parameters for the event buffer, such as maximum size, flush interval, and batch size. These settings control how events are stored and sent.

```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)
```
Tokens:
Raw