Metadata-Version: 2.4
Name: altissimo-mixpanel
Version: 0.1.1
Summary: Reusable Mixpanel analytics client for Altissimo projects
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: mixpanel,analytics,altissimo
Author: Altissimo HQ
Requires-Python: >=3.11
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Provides-Extra: mixpanel
Requires-Dist: mixpanel (>=4.0,<6) ; extra == "mixpanel"
Project-URL: Changelog, https://github.com/altissimo-hq/mixpanel-python/blob/main/CHANGELOG.md
Project-URL: Homepage, https://github.com/altissimo-hq/mixpanel-python
Project-URL: Issues, https://github.com/altissimo-hq/mixpanel-python/issues
Project-URL: Repository, https://github.com/altissimo-hq/mixpanel-python
Description-Content-Type: text/markdown

# altissimo-mixpanel

Reusable, typed Mixpanel analytics client for Altissimo projects.

## Features

- **Typed API** — full type annotations for all operations
- **Lazy SDK import** — `mixpanel` is imported only when needed, zero import-time side effects
- **Disabled mode** — no-op all calls when analytics is turned off or isn't configured
- **Error handling** — swallow-and-log by default, configurable via `on_error`
- **Protocol for DI** — `MixpanelClientProtocol` for dependency injection
- **Test helpers** — `FakeMixpanelClient` records all calls in-memory
- **Optional dependency** — works without `mixpanel` installed (fails gracefully)

## Installation

Published on PyPI as [`altissimo-mixpanel`](https://pypi.org/project/altissimo-mixpanel/);
the import name is `altissimo.mixpanel`.

```bash
# With the Mixpanel SDK (recommended)
pip install "altissimo-mixpanel[mixpanel]"

# Without the SDK (for type checking / protocol-only usage)
pip install altissimo-mixpanel
```

## Quick Start

### Using `from_env()`

```python
from altissimo.mixpanel import MixpanelClient

# Reads MIXPANEL_TOKEN from environment
# Missing token → disabled mode, with a warning (see Disabled Mode)
client = MixpanelClient.from_env()

# Track an event
client.track("user-123", "Page Viewed", properties={"page": "/home"})

# Set user profile properties
client.people_set("user-123", {"$name": "Jane Doe", "$email": "jane@example.com"})
```

### Direct Construction

```python
client = MixpanelClient(
    token="your-project-token",
    enabled=True,
    on_error="log",  # "log" | "raise" | "ignore"
)
```

### Disabled Mode

```python
# Explicitly disabled
client = MixpanelClient(token="...", enabled=False)
client.track("user-123", "Page Viewed")  # silently does nothing

# Via environment variable
# MIXPANEL_ENABLED=false → disabled
client = MixpanelClient.from_env()

# Missing token → automatically disabled, and a warning names the variable
# (no MIXPANEL_TOKEN set)
client = MixpanelClient.from_env()
assert not client.enabled
```

When the token is missing, `from_env()` behaves according to how tracking was
turned on:

| `enabled` | Missing or empty token |
| --- | --- |
| `True` (passed explicitly) | Raises `ValueError` naming the variable |
| `None` (default; `MIXPANEL_ENABLED` unset or truthy) | Disabled, one warning naming the variable |
| `False`, or `MIXPANEL_ENABLED` falsey | Disabled, no warning |

Pass `enabled=True` where a deployment must send events, so a missing secret
fails at startup instead of looking like Mixpanel going quiet. An app that only
sends events when deployed can pass that condition, e.g.
`MixpanelClient.from_env(enabled=IS_DEPLOYED)`: quiet locally, an error on a
deployment missing its token.

### Error Handling

```python
# Default: catch and log errors (analytics never crashes your app)
client = MixpanelClient(token="...", on_error="log")

# Raise errors (useful for debugging)
client = MixpanelClient(token="...", on_error="raise")

# Silently ignore errors
client = MixpanelClient(token="...", on_error="ignore")
```

### Custom Consumer

```python
from mixpanel import BufferedConsumer

# Use a buffered consumer for high-throughput
client = MixpanelClient(token="...", consumer=BufferedConsumer())
```

A `BufferedConsumer` only sends when its buffer fills, so flush it before the
process exits (a Cloud Run instance shutting down, a job or CLI command
finishing), or the buffered events are lost:

```python
client.flush()

# Or flush automatically when the block ends, even if it raises:
with MixpanelClient(token="...", consumer=BufferedConsumer()) as client:
    client.track("user-123", "Job Finished")
```

With the default consumer, which sends each message immediately, `flush()`
does nothing. A failing flush follows `on_error` like every other call.

Any object with `send(endpoint, json_message)` works as a consumer; the
`MixpanelConsumer` protocol describes it for type checking.

### Threads

One `MixpanelClient` can be shared across threads, such as a module-level
client in a web app. Each thread gets its own Mixpanel SDK instance and, with
the default consumer, its own HTTP session: the SDK's `Consumer` and
`BufferedConsumer` aren't safe to share between threads.

A `consumer=` is a single object every thread shares, so in a threaded app pass
a `consumer_factory` instead; it's called once per thread:

```python
from mixpanel import BufferedConsumer, Consumer

# A shorter request timeout, one consumer per thread
client = MixpanelClient.from_env(consumer_factory=lambda: Consumer(request_timeout=5))

# Buffered, one buffer per thread
client = MixpanelClient(token="...", consumer_factory=BufferedConsumer)
```

`flush()` flushes every thread's consumer. Call it when no other thread is
tracking, such as at shutdown.

### Testing with `FakeMixpanelClient`

```python
from altissimo.mixpanel import FakeMixpanelClient

def test_order_tracking():
    fake = FakeMixpanelClient()
    service = OrderService(mixpanel=fake)

    service.create_order(user_id="user-123", total=99.99)

    assert len(fake.tracked) == 1
    assert fake.tracked[0][1] == "Order Created"
    assert fake.tracked[0][2]["total"] == 99.99
```

### Protocol-Based Dependency Injection

```python
from altissimo.mixpanel import MixpanelClientProtocol

class OrderService:
    def __init__(self, mixpanel: MixpanelClientProtocol) -> None:
        self._mp = mixpanel

    def create_order(self, user_id: str, total: float) -> None:
        # ... business logic ...
        self._mp.track(user_id, "Order Created", properties={"total": total})
```

## API Reference

### `MixpanelClient`

| Method | Description |
|---|---|
| `track(distinct_id, event_name, *, properties?, meta?)` | Track an event |
| `people_set(distinct_id, properties, *, meta?)` | Set user profile properties |
| `people_set_once(distinct_id, properties, *, meta?)` | Set properties only if not set |
| `people_increment(distinct_id, properties, *, meta?)` | Increment numeric properties |
| `people_union(distinct_id, properties, *, meta?)` | Append to list properties |
| `people_unset(distinct_id, property_names, *, meta?)` | Remove properties |
| `alias(alias_id, original_id, *, meta?)` | Create an identity alias |

### Environment Variables

| Variable | Default | Description |
|---|---|---|
| `MIXPANEL_TOKEN` | — | Mixpanel project token |
| `MIXPANEL_ENABLED` | `true` | Set to `false`/`0`/`no` to disable |

## Development

```bash
# Install dependencies
poetry sync

# Run tests
poetry run pytest

# Run tests with coverage
poetry run pytest --cov=altissimo --cov-report=term-missing

# Lint
poetry run ruff check .

# Format
poetry run ruff format .
```

## License

[Apache-2.0](LICENSE)

