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:

pip install embrasure-analytics==0.1.0

From a source checkout:

pip install ./packages/analytics-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:

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:

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:

• 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.

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.

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

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:

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.