Metadata-Version: 2.5
Name: embrasure-analytics
Version: 0.1.0
Summary: Python event collection SDK for Embrasure
Project-URL: Homepage, https://embrasure.ai
Project-URL: Repository, https://github.com/EmbrasureAI/embrasure
License-Expression: MIT
License-File: LICENSE
License-File: NOTICE
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.27
Requires-Dist: typing-extensions<5,>=4.12
Description-Content-Type: text/markdown

# Embrasure analytics for Python

A typed Python 3.10+ SDK for the same native collection API used by
`@embrasure/analytics` and `@embrasure/analytics/server`. One client can serve
concurrent requests. Identity and groups are explicit on each event.

Install the package:

```sh
pip install embrasure-analytics==0.1.0
```

From a source checkout:

```sh
pip install ./packages/analytics-python
```

```python
import os
from embrasure_analytics import Client

analytics = Client(
    os.environ["EMBRASURE_ANALYTICS_PROJECT_KEY"],
    server_key=os.environ["EMBRASURE_ANALYTICS_SERVER_KEY"],
    host="https://api.embrasure.ai/collect",
)

analytics.track(
    "invoice_paid",
    user_id="user_123",
    id="invoice_paid:invoice_123",  # Reuse only for retries of this action.
    properties={"amount": 42, "currency": "USD"},
    groups={"company": "company_456"},
)
analytics.identify(
    "user_123",
    anonymous_id="browser_anonymous_id",  # Obtain from your browser integration.
    traits={"plan": "pro"},
    set_once={"signup_source": "pricing"},
)
analytics.group("company", "company_456", traits={"seats": 5})
result = analytics.shutdown(timeout=30)
# Inspect accepted, discarded, dropped, pending; acceptance precedes warehouse visibility.
```

`track(event, **fields)`, `identify(user_id, **fields)`, and
`group(group_type, group_key, **fields)` return whether a message was queued.
This is not a delivery receipt. Invalid messages return `False` and emit a
count-only diagnostic; invalid client configuration raises `ValueError`.

Common fields are `id`, `timestamp`, `anonymous_id`, `session_id`, `properties`,
`context`, `traits`, `set_once`, and `groups`. Track and group also accept
`user_id`. IDs are nonblank strings of at most 200 characters. Track requires a
user or anonymous ID; identify requires a user ID. Group may omit actor identity.
Custom event names cannot start with `$`. Unknown fields are rejected.

IDs and timestamps default to a UUID and UTC time. Timestamps accept a
timezone-aware `datetime` or ISO 8601 string and normalize to UTC. Properties
must be JSON values with string object keys; NaN, infinity, objects, cycles,
and excessive nesting are rejected. Serialization snapshots input immediately;
mutating caller dictionaries afterwards cannot alter queued events or retries.
Identify and group emit immutable messages. They do not establish a current
user, session, or company on the client. Trait reconstruction and identity
resolution happen in the analytics processor.

## Authentication and lifecycle

The public project key belongs in the body. A separate server credential goes
only in the Authorization header, just like the Node server SDK. `server_key`
is a required keyword: passing `None` explicitly chooses untrusted public-key
collection, matching browser/PostHog submissions. Never put a secret in the
project key. Analytics user IDs are attribution, not authenticated identity.

Keep one client per application process. A daemon worker starts on the first
valid event. Capture calls do no network I/O. HTTP connections are pooled;
redirects are not followed, and transport retries do not multiply SDK retries.

- `flush(timeout=10)` requests immediate delivery of work queued when called.
  It respects backoff and returns pending work after a failed attempt, as Node
  does. It can also return pending work if the total wait budget expires.
- `shutdown(timeout=30)` stops new events and drains through retries within one
  total wait budget. At the deadline, unsent events are counted as dropped.
  An already admitted HTTP request can finish afterwards and is reported pending.
  No new requests are admitted after shutdown stops the worker.
- `destroy()` drops buffered work immediately, matching Node's destroy behavior.
- A context manager calls shutdown on exit. Normal interpreter exit attempts a
  two-second drain. SIGKILL, crashes, and platform termination can still lose data.
- After a process fork, the child gets fresh locks, an empty queue, and a fresh
  HTTP connection pool. Only the parent delivers its inherited buffered events.
  Prefer creating clients in worker startup hooks with Gunicorn/preloaded apps.

`FlushResult` is an immutable object with `accepted`, `discarded`, `dropped`, and
`pending`. Method results report completions during that call; `statistics`
reports cumulative counts in this process, including earlier background sends
and local validation drops. Pending includes both queued and in-flight work.
Concurrent flush results may overlap; use statistics or diagnostics for metrics.

The optional `on_diagnostic(Diagnostic)` callback receives outcome, count, and a
fixed reason, never event bodies or secrets. Callback exceptions are isolated.
Callbacks may run on producer or worker threads, so keep them fast and
thread-safe. Lifecycle calls from a worker callback cannot wait on that worker;
shutdown/destroy in a callback drop remaining buffered work immediately.

## Delivery contract

Defaults deliberately match the JS/Node SDK:

| Behavior | Limit/default |
| --- | --- |
| Queued plus in-flight events | 500 |
| Flush interval / threshold | 5 seconds / 50 events |
| Batch size including JSON envelope | 60 KiB, at most 50 events |
| Event size including collector defaults | 32 KiB UTF-8 |
| JSON depth / groups | 10 levels / 20 groups |
| HTTP I/O timeout | 5 seconds per network phase |
| Delivery attempts | 5 total per event |
| Retry backoff | Exponential, plus jitter; Retry-After up to 5 minutes |

`flush_at` (1–50), `max_queue_size` (1–500), `flush_interval` (at least 0.1s),
and `request_timeout` are configurable. Retries cover network errors, 408, 429,
5xx, and ambiguous/malformed success receipts. Other HTTP errors drop the batch.
A valid success receipt must name `accepted` or `discarded` and acknowledge the
exact submitted count. An HTTP 2xx alone does not count as success.

The collector acknowledges native batches with HTTP 202 after sink acceptance;
that does not prove warehouse processing. Discard mode is reported separately.
The Flow writer deduplicates on workspace/project/environment/event ID, keeping
the first stored value. This is best-effort, memory-buffered analytics; use a
transactional outbox when an event must survive an application crash. There is
no automatic exception/PII capture, browser persistence, replay, or flag client.

## FastAPI, Django, and jobs

For FastAPI, track normally on the request thread/event loop; offload blocking
shutdown during application teardown:

```python
import asyncio
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from embrasure_analytics import Client


@asynccontextmanager
async def lifespan(app):
    analytics = Client(
        os.environ["EMBRASURE_ANALYTICS_PROJECT_KEY"],
        server_key=os.environ["EMBRASURE_ANALYTICS_SERVER_KEY"],
    )
    app.state.analytics = analytics
    try:
        yield
    finally:
        await asyncio.to_thread(analytics.shutdown, timeout=10)


app = FastAPI(lifespan=lifespan)
# In a route: request.app.state.analytics.track("report_created", user_id=user.id)
```

For Django, initialize one client in each serving worker, then call
`analytics.track("report_created", user_id=str(request.user.pk))` after successful
work. Use the process server's worker-exit hook to call shutdown. Do not keep the
current request's user on a module global. No framework dependency is installed.

For scripts, jobs, and serverless invocations, use `with Client(...) as analytics`
or explicitly call `shutdown(timeout=...)` within the platform's remaining time.
Create a new client after shutdown; a shut-down client cannot be reused. The
context manager's default drain budget is 30 seconds.

## PostHog prior art and compatibility

Reviewed and adapted from **PostHog Python 7.51.0**, source revision
[`f6c4b05865974d0945499c33d548d1099f9945bb`](https://github.com/PostHog/posthog-python/tree/f6c4b05865974d0945499c33d548d1099f9945bb):

- `posthog/client.py`: immutable enqueue, explicit lifecycle, weak fork/exit
  hooks, and protection against waiting on the consumer from callbacks.
- `posthog/consumer.py`: daemon batching by count/time/bytes, bounded delivery,
  retry classification, and releasing queue capacity after completion.
- `posthog/request.py` and `capture_v1.py`: pooled transport, Retry-After parsing,
  and retry backoff. Embrasure handles retries at one layer and validates receipts.
- Upstream `test/test_consumer.py`: shutdown admission, queue acknowledgement,
  payload sizing, callback failures, and retry cases informed our tests.

Upstream defaults allow much larger messages/queues and use a different wire
format. We retain Embrasure's lower JS/Node limits, native messages and explicit
receipt outcomes. HTTPX provides transport; PostHog is pinned as a **test-only**
dependency. No runtime fork, monkey patch, PostHog account, or PostHog service is
required. Upstream MIT attribution is included in [NOTICE](NOTICE).

The real pinned PostHog Python SDK is also tested against the collector for
capture, person set/set_once, groups, page events, and gzip. Tests found and
fixed its requirement for HTTP **200** and its separate `$set_once` event.
Native clients continue to use `/v1/batch` and HTTP 202. PostHog compatibility
covers capture, person set/set_once, groups, page events, and gzip with the v0
capture protocol. Feature flags, AI/error capture, alias, historical import, and
the v1 capture protocol are outside this integration.

## Verification

The following commands and examples are for the Embrasure monorepo checkout;
the public package does not include the collector or internal test fixtures.

```sh
uv sync --project apps/api --frozen --dev
uv sync --project packages/analytics-python --frozen
uv run --project packages/analytics-python --frozen pytest -q \
  -c packages/analytics-python/pyproject.toml packages/analytics-python/tests
uv run --project packages/analytics-python --frozen ruff check packages/analytics-python
uv run --project packages/analytics-python --frozen mypy \
  --config-file packages/analytics-python/pyproject.toml packages/analytics-python/src
uv build --project packages/analytics-python
```

CI installs the built wheel and runs the tests on Python 3.10 and 3.14. The
shared `packages/analytics/tests/native-contract.json` fixture also runs through
the Node SDK and real collector. Unit tests exercise failures, concurrency,
limits, lifecycle deadlines and actual POSIX forks. Integration tests exercise
authenticated native HTTP, public attribution, rejected keys, real PostHog calls,
and interpreter-exit delivery. A recording sink does not prove warehouse storage.

For the full warehouse path, install the wheel in a clean environment and run
[examples/warehouse_smoke.py](examples/warehouse_smoke.py) with a configured
synthetic project and its **actual destination workspace**. The public
`pk_collection_smoke` key is operator-configured in the internal workspace; it
cannot select another workspace through SDK fields. The script checks live
collector health, sends 66 originals, two altered retries and four ordering
markers, and records the inputs, delivery counts and scoped verification SQL.
Run that SQL against the destination warehouse through an authorized product
session/MCP. Save its rows as JSON and verify them:

```sh
python packages/analytics-python/examples/warehouse_smoke.py \
  --workspace <configured-smoke-workspace> --manifest /tmp/python-sdk-smoke.json
python packages/analytics-python/examples/warehouse_smoke.py \
  --manifest /tmp/python-sdk-smoke.json --verify-rows /tmp/warehouse-rows.json
```

Verification requires exactly 70 unique persisted rows, original properties
surviving the changed retries, correct timestamps, identity, group/trait data,
trust metadata, and Unicode/nested JSON. Read the full result page; acceptance
alone cannot satisfy this check. No customer events or production configuration
are changed by the smoke script.

For a complete local run, start a dedicated copy of the private Flow repository's
`tests/local/compose.yaml` fixture with its Kafka profile and dynamic ports (see
that repository's `tests/local/README.md`). Then run the checked-in driver:

```sh
uv run --with ./packages/analytics-python/dist/embrasure_analytics-0.1.0-py3-none-any.whl \
  packages/analytics-python/examples/local_warehouse_smoke.py \
  --binary /path/to/embrasure-flow-internal/target/release/embrasure-flow \
  --kafka 127.0.0.1:<kafka-port> \
  --catalog-uri http://127.0.0.1:<rest-port> \
  --s3-endpoint http://127.0.0.1:<minio-port> \
  --artifacts /tmp/python-sdk-warehouse-new-run
```

This requires the API's local `.venv` and a new artifact directory. The driver
uses a unique topic, consumer group and namespace, copies the binary for a
stable run, uses the production analytics column mapping, and starts only its
own collector/writer processes. It verifies trusted native events, altered
retries, SIGKILL/restart with retained state, and five real PostHog messages via
stock DuckDB `iceberg_scan`. The caller owns starting/stopping the disposable
Docker services. Local and live end-to-end results are recorded in
[VALIDATION.md](VALIDATION.md).
