diff --git a/.github/workflows/sdk-compliance.yml b/.github/workflows/sdk-compliance.yml index 1dfcb908..be989753 100644 --- a/.github/workflows/sdk-compliance.yml +++ b/.github/workflows/sdk-compliance.yml @@ -13,9 +13,19 @@ on: jobs: compliance: - name: PostHog SDK compliance tests + name: PostHog SDK compliance tests (capture v0) uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@02c049e529001d02f37a534745678e057d371fb0 with: adapter-dockerfile: "sdk_compliance_adapter/Dockerfile" adapter-context: "." test-harness-version: "0.10.0" + report-name: "sdk-compliance-report-v0" + + compliance-v1: + name: PostHog SDK compliance tests (capture v1) + uses: PostHog/posthog-sdk-test-harness/.github/workflows/test-sdk-action.yml@02c049e529001d02f37a534745678e057d371fb0 + with: + adapter-dockerfile: "sdk_compliance_adapter/Dockerfile.v1" + adapter-context: "." + test-harness-version: "0.10.0" + report-name: "sdk-compliance-report-v1" diff --git a/.sampo/changesets/capture-v1-mode.md b/.sampo/changesets/capture-v1-mode.md new file mode 100644 index 00000000..cffd1471 --- /dev/null +++ b/.sampo/changesets/capture-v1-mode.md @@ -0,0 +1,9 @@ +--- +pypi/posthog: minor +--- + +Add an opt-in `capture_mode` for the Capture V1 ingestion protocol (`POST /i/v1/analytics/events`). Set `capture_mode="v1"` on the client (or the `POSTHOG_CAPTURE_MODE=v1` environment variable) to use Bearer auth, per-event results, and partial retry. Defaults to `"v0"` (the legacy `/batch/` endpoint), so existing setups are unaffected. + +When using `capture_mode="v1"`, request bodies can be compressed via `capture_compression` (or `POSTHOG_CAPTURE_COMPRESSION`): `"gzip"`, `"deflate"`, `"zstd"` (requires the optional `posthog[zstd]` extra), or `"none"` (default). The legacy `gzip=True` flag is honored as a fallback. + +Per-event server verdicts are surfaced through the existing `on_error` handler: events the backend explicitly drops, or fails to accept after retries, raise a `CaptureV1Error` carrying the affected event UUIDs — so a rejection is never silently lost, even when the HTTP request itself succeeded. diff --git a/AGENTS.md b/AGENTS.md index 8a53f11d..7e0af378 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -9,6 +9,27 @@ Guidance for coding agents working in `posthog-python`. - The project uses `uv` for local development. See `CONTRIBUTING.md` for setup. - Keep edits targeted and follow existing patterns. Prefer adding or updating tests near the behavior you change. +## Capture protocol (`capture_mode`) + +The client supports two ingestion wire protocols, selected by `capture_mode` (precedence: explicit `Client(capture_mode=...)` kwarg > `POSTHOG_CAPTURE_MODE` env var > default). + +- `"v0"` (default) — legacy `POST /batch/`. Upgrades stay transparent; existing callers are unaffected. +- `"v1"` — `POST /i/v1/analytics/events`: Bearer auth, a typed event `options` object, per-event results, and partial retry. + +v1 request bodies can additionally be compressed via `capture_compression` (precedence: explicit `Client(capture_compression=...)` kwarg > `POSTHOG_CAPTURE_COMPRESSION` env var > the legacy `gzip` flag > none). Supported values are `"none"`, `"gzip"`, `"deflate"` (zlib-wrapped, RFC 1950, to match the server's decoder and the Go/Rust SDKs), and `"zstd"` (requires the optional `posthog[zstd]` extra; explicit zstd without the package raises, env-var zstd warns and falls back). v0 keeps using its own `gzip` flag; `capture_compression` is v1-only. + +Where the pieces live: + +- `posthog/capture_mode.py` — the `CaptureMode` enum and `_resolve_capture_mode()` precedence logic. +- `posthog/capture_compression.py` — the `CaptureCompression` enum and `_resolve_capture_compression()` precedence logic (with `gzip` fallback). +- `posthog/capture_v1.py` — pure transforms (`_to_v1_event`, `_build_v1_batch_body`) and transport (`_post_v1`, `_compress_v1`, `_parse_v1_response`, `_send_v1_batch`, `CaptureV1Error`). +- Public API surface (enforced by `references/public_api_snapshot.txt`): `CaptureMode`, `CaptureCompression` (both re-exported from `posthog`), `CaptureV1Error`, and the two env var names. Everything else in these modules is underscore-private plumbing. +- Routing: `Consumer._send_analytics` (async) and `Client._enqueue` (sync) pick the analytics submitter by `capture_mode`. The dedicated `$ai_*` endpoint has no v1 form and always uses the legacy submitter. + +v1-specific behavior to preserve when editing: sentinel `$`-properties are lifted into `options` (coerced to native JSON types or omitted — a wrong type 400s the whole batch); top-level `$set`/`$set_once` are relocated into `properties`; only events the server tags `retry` are resent (stable `PostHog-Request-Id`/`created_at`, incrementing `PostHog-Attempt`); a server `drop` is a terminal per-event rejection — drops are accumulated across attempts and surfaced via `CaptureV1Error`/`on_error` even on a 2xx with no retries (a success status is not full delivery); `Retry-After` is a *minimum*, not a replacement (the client waits `max(configured_backoff, min(Retry-After, _MAX_BACKOFF_SECONDS))`); `_MAX_BACKOFF_SECONDS` (30s) is the single ceiling for both the exponential backoff and the `Retry-After` clamp; `429` is terminal. + +Retry blocking matches v0: in the default async mode retries happen on the background consumer thread, but with `sync_mode=True` the partial-retry loop (including its backoff sleeps) runs inline on the calling thread, so a slow/erroring endpoint blocks the caller until retries are exhausted. + ## Validation Useful checks: diff --git a/posthog/__init__.py b/posthog/__init__.py index 2184d372..05002962 100644 --- a/posthog/__init__.py +++ b/posthog/__init__.py @@ -9,6 +9,8 @@ OptionalCaptureArgs, OptionalSetArgs, ) +from posthog.capture_compression import CaptureCompression as CaptureCompression +from posthog.capture_mode import CaptureMode as CaptureMode from posthog.client import Client from posthog.exception_capture import ExceptionCapture from posthog.contexts import ( @@ -381,6 +383,9 @@ def get_tags() -> Dict[str, Any]: # We recommend setting this to False if you are only using the personalApiKey for evaluating remote config payloads via `get_remote_config_payload` and not using local evaluation. enable_local_evaluation = True # type: bool flag_definition_cache_provider = None # type: Optional[FlagDefinitionCacheProvider] +# Capture wire protocol for the global client. None defers to POSTHOG_CAPTURE_MODE +# then CaptureMode.V0. See posthog.capture_mode.CaptureMode. +capture_mode = None # type: Optional[CaptureMode] default_client = None # type: Optional[Client] @@ -1180,6 +1185,7 @@ def setup() -> Client: exception_autocapture_bucket_size=exception_autocapture_bucket_size, exception_autocapture_refill_rate=exception_autocapture_refill_rate, exception_autocapture_refill_interval_seconds=exception_autocapture_refill_interval_seconds, + capture_mode=capture_mode, ) # Always set in case user changes it. Preserve Client's auto-disabled state diff --git a/posthog/capture_compression.py b/posthog/capture_compression.py new file mode 100644 index 00000000..659c12fa --- /dev/null +++ b/posthog/capture_compression.py @@ -0,0 +1,128 @@ +import logging +import os +from enum import Enum +from typing import Any, Optional, Union + +_zstandard: Any | None +try: + import zstandard + + _zstandard = zstandard +except ImportError: + _zstandard = None + +__all__ = ["CAPTURE_COMPRESSION_ENV_VAR", "CaptureCompression"] + +log = logging.getLogger("posthog") + +CAPTURE_COMPRESSION_ENV_VAR = "POSTHOG_CAPTURE_COMPRESSION" + + +class CaptureCompression(str, Enum): + """Selects the request-body compression for capture-v1 uploads. + + Only honored when ``capture_mode`` is ``V1``; the legacy ``/batch/`` path + keeps using its own ``gzip`` flag. ``NONE`` sends the body uncompressed. + ``GZIP`` and ``DEFLATE`` (zlib, RFC 1950) are both stdlib / zero-dependency; + ``ZSTD`` is faster and compresses better but needs the optional zstandard + package (``pip install posthog[zstd]``) until stdlib support lands in + Python 3.14. Each maps to the matching ``Content-Encoding`` token the v1 + server decodes (``br`` is accepted by the server too but is intentionally + left out for now). Inheriting from ``str`` keeps the members comparable to + and serializable as their token values. + """ + + NONE = "none" + GZIP = "gzip" + DEFLATE = "deflate" + ZSTD = "zstd" + + +# Accepted spellings for both the kwarg and the env var. ``identity`` mirrors +# the HTTP token for "no encoding". +_ALIASES: dict[str, CaptureCompression] = { + "none": CaptureCompression.NONE, + "identity": CaptureCompression.NONE, + "gzip": CaptureCompression.GZIP, + "deflate": CaptureCompression.DEFLATE, + "zstd": CaptureCompression.ZSTD, +} + + +def _zstd_available() -> bool: + return _zstandard is not None + + +def _coerce_explicit( + value: Union[CaptureCompression, str], +) -> CaptureCompression: + """Normalize an explicitly-supplied compression to a ``CaptureCompression``. + + An explicit but unrecognized value is a programming error, so it raises + ``ValueError`` rather than silently defaulting (unlike the env var, which is + operator-supplied and defaults defensively). + """ + if isinstance(value, CaptureCompression): + return value + if isinstance(value, str): + resolved = _ALIASES.get(value.strip().lower()) + if resolved is not None: + return resolved + raise ValueError( + f"invalid capture_compression {value!r}; expected a CaptureCompression " + f"or one of {sorted(_ALIASES)}" + ) + + +def _resolve_capture_compression( + capture_compression: Optional[Union[CaptureCompression, str]] = None, + *, + gzip_fallback: bool = False, +) -> CaptureCompression: + """Resolve the effective v1 compression. + + Precedence: explicit ``capture_compression`` argument > + ``POSTHOG_CAPTURE_COMPRESSION`` env var > the legacy ``gzip`` flag + (``GZIP`` when set) > ``NONE``. An unrecognized env value logs a warning and + falls back to the ``gzip`` flag, so a typo never silently changes encoding. + + ``ZSTD`` requires the optional zstandard package: explicitly requesting it + without the package raises ``ValueError`` (programming error, fail loud), + while requesting it via the env var warns and falls back (operator-supplied + config must never silently break capture). + """ + if capture_compression is not None: + resolved = _coerce_explicit(capture_compression) + if resolved is CaptureCompression.ZSTD and not _zstd_available(): + raise ValueError( + "capture_compression 'zstd' requires the zstandard package; " + "install posthog[zstd]" + ) + return resolved + + fallback = CaptureCompression.GZIP if gzip_fallback else CaptureCompression.NONE + + raw = os.environ.get(CAPTURE_COMPRESSION_ENV_VAR) + if raw is None or raw.strip() == "": + return fallback + + env_resolved = _ALIASES.get(raw.strip().lower()) + if env_resolved is None: + log.warning( + "Unrecognized %s=%r; falling back to %s. Expected one of %s.", + CAPTURE_COMPRESSION_ENV_VAR, + raw, + fallback.value, + sorted(_ALIASES), + ) + return fallback + if env_resolved is CaptureCompression.ZSTD and not _zstd_available(): + log.warning( + "%s=%r requires the zstandard package (install posthog[zstd]); " + "falling back to %s.", + CAPTURE_COMPRESSION_ENV_VAR, + raw, + fallback.value, + ) + return fallback + return env_resolved diff --git a/posthog/capture_mode.py b/posthog/capture_mode.py new file mode 100644 index 00000000..7ea3696f --- /dev/null +++ b/posthog/capture_mode.py @@ -0,0 +1,84 @@ +import logging +import os +from enum import Enum +from typing import Optional, Union + +__all__ = ["CAPTURE_MODE_ENV_VAR", "CaptureMode"] + +log = logging.getLogger("posthog") + +CAPTURE_MODE_ENV_VAR = "POSTHOG_CAPTURE_MODE" + + +class CaptureMode(str, Enum): + """Selects the capture wire protocol used for event ingestion. + + ``V0`` is the legacy ``POST /batch/`` endpoint and the default, so upgrading + is transparent to existing callers. ``V1`` opts into + ``POST /i/v1/analytics/events`` (Bearer auth, per-event results, partial + retry). Inheriting from ``str`` keeps the members directly comparable to and + serializable as their ``"v0"`` / ``"v1"`` values. + """ + + V0 = "v0" + V1 = "v1" + + +# Accepted spellings for both the explicit kwarg and the env var. Aliases mirror +# the posthog-go naming (``legacy`` / ``analytics_v1``) so the two SDKs are +# configured with the same vocabulary. +_ALIASES: dict[str, CaptureMode] = { + "v0": CaptureMode.V0, + "legacy": CaptureMode.V0, + "v1": CaptureMode.V1, + "analytics_v1": CaptureMode.V1, +} + + +def _coerce_explicit(value: Union[CaptureMode, str]) -> CaptureMode: + """Normalize an explicitly-supplied capture mode to a ``CaptureMode``. + + Accepts a ``CaptureMode`` or one of the string aliases. An explicit but + unrecognized value is a programming error, so it raises ``ValueError`` rather + than silently defaulting (unlike the env var, which is operator-supplied and + defaults defensively). + """ + if isinstance(value, CaptureMode): + return value + if isinstance(value, str): + resolved = _ALIASES.get(value.strip().lower()) + if resolved is not None: + return resolved + raise ValueError( + f"invalid capture_mode {value!r}; expected a CaptureMode or one of " + f"{sorted(_ALIASES)}" + ) + + +def _resolve_capture_mode( + capture_mode: Optional[Union[CaptureMode, str]] = None, +) -> CaptureMode: + """Resolve the effective capture mode. + + Precedence: explicit ``capture_mode`` argument > ``POSTHOG_CAPTURE_MODE`` env + var > ``CaptureMode.V0``. An unrecognized env value logs a warning and falls + back to ``V0`` so a typo never silently flips the wire protocol. + """ + if capture_mode is not None: + return _coerce_explicit(capture_mode) + + raw = os.environ.get(CAPTURE_MODE_ENV_VAR) + if raw is None or raw.strip() == "": + return CaptureMode.V0 + + resolved = _ALIASES.get(raw.strip().lower()) + if resolved is None: + log.warning( + "Unrecognized %s=%r; falling back to %s. Expected one of %s.", + CAPTURE_MODE_ENV_VAR, + raw, + CaptureMode.V0.value, + sorted(_ALIASES), + ) + return CaptureMode.V0 + return resolved diff --git a/posthog/capture_v1.py b/posthog/capture_v1.py new file mode 100644 index 00000000..a4f199dd --- /dev/null +++ b/posthog/capture_v1.py @@ -0,0 +1,612 @@ +"""Serialization and transport for the Capture V1 wire protocol. + +This module owns everything specific to ``POST /i/v1/analytics/events``: the +*transform* layer (legacy-shaped queued message -> v1 wire event + batch +envelope) and the *transport* layer (a single HTTP attempt, response parsing, +and the partial-retry send loop). + +The v1 contract (see ``rust/capture/src/v1/analytics/types.rs``) differs from +the legacy ``/batch/`` shape in a few load-bearing ways that this module +encodes: + +- A typed ``options`` object carries a handful of sentinel properties, renamed + and strictly typed. Wrong JSON types fail deserialization of the *whole + batch*, so values are coerced to native types or omitted entirely. +- ``$set``/``$set_once`` have no top-level form in v1; the server reads them + from ``properties``. The legacy ``set()``/``set_once()`` builders emit them at + the top level, so they are relocated into ``properties`` here. +- ``$lib``/``$lib_version`` are injected server-side from the required + ``PostHog-Sdk-Info`` header and are stripped from v1 properties. + +The response is per-event: a 200 carries a ``results`` map keyed by event uuid, +each tagged ``ok``/``warning`` (terminal-success), ``drop`` (terminal-failure), +or ``retry``. :func:`_send_v1_batch` resends only the ``retry`` events on the next +attempt, holding the ``PostHog-Request-Id`` and batch ``created_at`` stable +across attempts while incrementing ``PostHog-Attempt``. ``ok``/``warning``/absent +events succeed; ``drop`` and retry-exhaustion are carried on the +:class:`CaptureV1Error` raised on batch-level/terminal failure, so the consumer's +existing ``on_error(exc, batch)`` path surfaces them unchanged (no per-event +logging of its own). + +Request bodies are optionally compressed per :class:`~posthog.capture_compression.CaptureCompression` +(``gzip`` or zlib-wrapped ``deflate``), advertised via ``Content-Encoding``. +""" + +import json +import logging +import time +import zlib +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime +from gzip import GzipFile +from io import BytesIO +from typing import TYPE_CHECKING, Any, Optional +from uuid import uuid4 + +from posthog.capture_compression import CaptureCompression, _zstandard +from posthog.request import ( + DatetimeSerializer, + USER_AGENT, + APIError, + _get_session, + normalize_host, +) +from posthog.utils import guess_timezone as _guess_timezone, remove_trailing_slash + +if TYPE_CHECKING: + import requests + +log = logging.getLogger("posthog") + +# Only the error type is public API: it reaches user code through `on_error` +# callbacks, so callers may want to catch/inspect it. Everything else is +# submitter plumbing. +__all__ = ["CaptureV1Error"] + +_CAPTURE_V1_PATH = "/i/v1/analytics/events" + +# Required request/response headers for the v1 endpoint. Defined here as the +# single source of truth; the transport layer builds requests from them. +_HEADER_SDK_INFO = "PostHog-Sdk-Info" +_HEADER_ATTEMPT = "PostHog-Attempt" +_HEADER_REQUEST_ID = "PostHog-Request-Id" +_HEADER_REQUEST_TIMESTAMP = "PostHog-Request-Timestamp" + +# Per-event result codes the backend emits (rust EventResult). `ok`/`warning` +# are terminal-success; `drop` terminal-failure; `retry` is safe to resend. +_RESULT_OK = "ok" +_RESULT_WARNING = "warning" +_RESULT_DROP = "drop" +_RESULT_RETRY = "retry" + +# HTTP status classification. 429 is terminal in v1 (unlike v0, where it is +# retried) — the backend signals overload via retryable 5xx + Retry-After. +_RETRYABLE_STATUSES = frozenset({408, 500, 502, 503, 504}) +_TERMINAL_STATUSES = frozenset({400, 401, 402, 413, 415, 429}) + +# Single ceiling (seconds) for the retry backoff: caps the exponential schedule +# and clamps a server ``Retry-After`` to the same value. Keeps the max retry +# wait bounded (a hostile/buggy header can't park the consumer thread) and +# unifies the default with posthog-go/posthog-rs (all 30s). +_MAX_BACKOFF_SECONDS = 30 + +# Sentinel properties lifted to top-level string fields on the event. +_TOPLEVEL_SENTINELS: tuple[tuple[str, str], ...] = ( + ("$session_id", "session_id"), + ("$window_id", "window_id"), +) + +# Top-level legacy keys relocated into properties (v1 has no top-level form). +_RELOCATE_TO_PROPERTIES = ("$set", "$set_once") + +# Properties dropped from v1 events (server injects them from PostHog-Sdk-Info). +_STRIP_FROM_PROPERTIES = ("$lib", "$lib_version") + + +def _coerce_bool(value: Any) -> Optional[bool]: + """Coerce a sentinel value to ``bool`` using the backend's truthiness rules. + + Native bool passes through; ``"true"``/``"1"`` and ``"false"``/``"0"`` + (case-insensitive, trimmed) map to the obvious bool; any other numeric value + is nonzero-truthy. Anything else returns ``None`` so the option is omitted + rather than sent with a type the strict v1 schema would reject. + """ + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in ("true", "1"): + return True + if normalized in ("false", "0"): + return False + return None + if isinstance(value, (int, float)): + return value != 0 + return None + + +def _coerce_str(value: Any) -> Optional[str]: + """Accept only ``str`` (the backend's ``product_tour_id`` is ``Option``).""" + return value if isinstance(value, str) else None + + +# Sentinel properties lifted into the typed `options` object: legacy property +# key, the backend's field name, and the coercer enforcing its strict type +# (wrong JSON types fail deserialization of the whole batch, so a value that +# won't coerce is omitted). The coercer is stored directly to keep the dispatch +# type-checked rather than keyed by a stringly-typed name. +_OPTION_SENTINELS: tuple[tuple[str, str, Callable[[Any], Any]], ...] = ( + ("$cookieless_mode", "cookieless_mode", _coerce_bool), + ("$ignore_sent_at", "disable_skew_correction", _coerce_bool), + ("$product_tour_id", "product_tour_id", _coerce_str), + ("$process_person_profile", "process_person_profile", _coerce_bool), +) + + +def _v1_timestamp(timestamp: Any) -> str: + """Return a timezone-aware RFC3339 timestamp string. + + Messages off the queue already carry an ISO-8601 string (``_enqueue`` runs + ``guess_timezone(...).isoformat()``), so that is passed through. A + ``datetime`` is normalized to timezone-aware and serialized; a missing value + defaults to now in UTC. The v1 server parses strictly with + ``DateTime::parse_from_rfc3339`` and rejects naive timestamps. + """ + if timestamp is None: + return datetime.now(timezone.utc).isoformat() + if isinstance(timestamp, datetime): + return _guess_timezone(timestamp).isoformat() + return timestamp + + +def _to_v1_event(msg: dict) -> dict: + """Transform a legacy-shaped queued message into a v1 wire event. + + Pure: the input ``msg`` is not mutated (a fresh ``properties`` dict is + built), so it remains safe to keep the original for retries or callbacks. + """ + properties = dict(msg.get("properties") or {}) + + # Relocate top-level $set/$set_once into properties; v1 has no top-level + # form. On the unusual collision where properties already carries the key, + # the properties value wins. + for key in _RELOCATE_TO_PROPERTIES: + top_val = msg.get(key) + if top_val is None: + continue + existing = properties.get(key) + if isinstance(top_val, dict) and isinstance(existing, dict): + properties[key] = {**top_val, **existing} + elif key not in properties: + properties[key] = top_val + + for key in _STRIP_FROM_PROPERTIES: + properties.pop(key, None) + + options: dict[str, Any] = {} + for prop_key, wire_key, coercer in _OPTION_SENTINELS: + if prop_key not in properties: + continue + # Always removed from properties — these sentinels must never reach v1 + # backend properties — but only emitted as an option when coercible. + coerced = coercer(properties.pop(prop_key)) + if coerced is not None: + options[wire_key] = coerced + + top_level: dict[str, str] = {} + for prop_key, field_name in _TOPLEVEL_SENTINELS: + if prop_key not in properties: + continue + coerced_str = _coerce_str(properties.pop(prop_key)) + if coerced_str is not None: + top_level[field_name] = coerced_str + + event = { + "event": msg["event"], + "uuid": msg["uuid"], + "distinct_id": msg["distinct_id"], + "timestamp": _v1_timestamp(msg.get("timestamp")), + # Always a dict so it serializes as "{}" rather than null when empty. + "options": options, + "properties": properties, + } + event.update(top_level) + return event + + +def _build_v1_batch_body( + events: list[dict], + historical_migration: bool = False, + created_at: Optional[str] = None, +) -> dict: + """Assemble the v1 batch envelope. + + Carries no ``api_key`` (Bearer auth) and no ``sent_at``. + ``historical_migration`` is omitted when False (the server defaults it). + ``created_at`` defaults to now in UTC; :func:`_send_v1_batch` passes a value + hoisted once so it stays stable across retry attempts. + """ + body: dict[str, Any] = { + "created_at": created_at or datetime.now(timezone.utc).isoformat(), + "batch": events, + } + if historical_migration: + body["historical_migration"] = True + return body + + +@dataclass +class _V1EventResult: + """A single event's directive from a 2xx ``results`` map.""" + + result: Optional[str] + details: Optional[str] = None + + +@dataclass +class _V1ParsedResponse: + """Classified outcome of one v1 HTTP attempt. + + ``is_success`` is the 2xx classification. On success ``results`` holds the + per-uuid directives (``None``/``malformed=True`` when the body could not be + parsed — treated as terminal so a bad success never loops forever). On a + non-2xx, ``error_message`` is the best-effort human-readable detail. + """ + + status_code: int + is_success: bool + retry_after: Optional[float] = None + results: Optional[dict[str, _V1EventResult]] = None + malformed: bool = False + error_message: str = "" + + +class CaptureV1Error(APIError): + """Batch-level failure of a capture-v1 send. + + Subclasses :class:`APIError` so the consumer's existing ``on_error`` handling + (which already inspects ``status``/``retry_after``) keeps working; the extra + fields carry v1 specifics for richer logging/callbacks. + """ + + def __init__( + self, + status: int | str, + message: str, + *, + retry_after: Optional[float] = None, + request_id: Optional[str] = None, + attempts: Optional[int] = None, + retry_exhausted: Optional[list[str]] = None, + drops: Optional[list[tuple[str, Optional[str]]]] = None, + ): + super().__init__(status, message, retry_after=retry_after) + self.request_id = request_id + self.attempts = attempts + # uuids the server told us to retry but we never delivered (exhausted). + self.retry_exhausted = retry_exhausted or [] + # (uuid, details) pairs the server told us to drop on a 2xx response. + self.drops = drops or [] + + +def _is_success_status(status: int) -> bool: + return 200 <= status < 300 + + +def _parse_retry_after(header_value: Optional[str]) -> Optional[float]: + """Parse a ``Retry-After`` header (delta-seconds or HTTP-date) to seconds.""" + if not header_value: + return None + try: + return float(header_value) + except (ValueError, TypeError): + pass + try: + delta = parsedate_to_datetime(header_value) - datetime.now(timezone.utc) + return max(0.0, delta.total_seconds()) + except (ValueError, TypeError): + return None + + +def _compress_v1( + compression: CaptureCompression, data: str +) -> tuple[str | bytes, Optional[str]]: + """Compress a v1 request body, returning ``(body, Content-Encoding token)``. + + ``GZIP`` emits a gzip stream; ``DEFLATE`` emits a *zlib-wrapped* deflate + stream (RFC 1950, leading ``0x78``) to match posthog-go / posthog-rs and the + server's zlib decoder for ``Content-Encoding: deflate`` — raw, headerless + deflate would be misrouted. ``ZSTD`` emits a standard zstd frame via the + optional zstandard package. ``NONE`` returns the string body and no token. + """ + if compression == CaptureCompression.GZIP: + buf = BytesIO() + with GzipFile(fileobj=buf, mode="w") as gz: + # `data` is produced by json.dumps(), whose default encoding is utf-8. + gz.write(data.encode("utf-8")) + return buf.getvalue(), "gzip" + if compression == CaptureCompression.DEFLATE: + return zlib.compress(data.encode("utf-8")), "deflate" + if compression == CaptureCompression.ZSTD: + # _resolve_capture_compression only yields ZSTD when zstandard is + # importable; this guard covers direct Consumer construction. + if _zstandard is None: + raise ValueError( + "capture_compression 'zstd' requires the zstandard package; " + "install posthog[zstd]" + ) + return _zstandard.ZstdCompressor().compress(data.encode("utf-8")), "zstd" + return data, None + + +def _post_v1( + api_key: str, + host: Optional[str], + batch_body: dict, + *, + attempt: int, + request_id: str, + compression: CaptureCompression = CaptureCompression.NONE, + timeout: int = 15, + session: Optional["requests.Session"] = None, +) -> "requests.Response": + """Perform a single ``POST /i/v1/analytics/events`` attempt. + + Bearer-authed (no ``api_key`` in the body) with the required v1 headers. + ``attempt`` (1-based) and the stable ``request_id`` are echoed via + ``PostHog-Attempt``/``PostHog-Request-Id`` so the backend can correlate + retries. The body is compressed per ``compression`` (advertised via + ``Content-Encoding``). Returns the raw response; classification is left to + the caller. + """ + trimmed_host = remove_trailing_slash(normalize_host(host)) + url = trimmed_host + _CAPTURE_V1_PATH + data = json.dumps(batch_body, cls=DatetimeSerializer) + headers = { + "Content-Type": "application/json", + "User-Agent": USER_AGENT, + "Authorization": f"Bearer {api_key}", + _HEADER_SDK_INFO: USER_AGENT, + _HEADER_ATTEMPT: str(attempt), + _HEADER_REQUEST_ID: request_id, + _HEADER_REQUEST_TIMESTAMP: datetime.now(timezone.utc).isoformat(), + } + body, encoding = _compress_v1(compression, data) + if encoding is not None: + headers["Content-Encoding"] = encoding + + log.debug("capture v1 POST %s attempt=%s request_id=%s", url, attempt, request_id) + return (session or _get_session()).post( + url, data=body, headers=headers, timeout=timeout + ) + + +def _parse_v1_response(res: "requests.Response") -> _V1ParsedResponse: + """Read and classify a v1 response without raising.""" + status = res.status_code + retry_after = _parse_retry_after(res.headers.get("Retry-After")) + + if _is_success_status(status): + try: + payload = res.json() + raw_results = payload["results"] + results = { + uid: _V1EventResult( + result=(r or {}).get("result"), + details=(r or {}).get("details"), + ) + for uid, r in raw_results.items() + } + return _V1ParsedResponse(status, True, retry_after, results=results) + except (ValueError, KeyError, AttributeError, TypeError): + # 2xx with a body we can't read as a results map: terminal, so we + # don't loop forever re-sending against a broken success. + return _V1ParsedResponse(status, True, retry_after, malformed=True) + + message = "" + try: + payload = res.json() + if isinstance(payload, dict): + message = ( + payload.get("error_description") + or payload.get("error") + or payload.get("detail") + or "" + ) + except (ValueError, AttributeError): + pass + if not message: + message = res.text or f"capture v1 request failed with status {status}" + return _V1ParsedResponse(status, False, retry_after, error_message=message) + + +def _backoff(attempt_index: int, retry_after: Optional[float]) -> None: + """Sleep before the next attempt. + + Exponential backoff capped at :data:`_MAX_BACKOFF_SECONDS` is the base. When + the server sent a ``Retry-After`` it acts as a *minimum*, not a replacement: + the client waits the longer of the configured backoff and ``Retry-After``, so + a small ``Retry-After`` never retries earlier than the normal schedule + (matching posthog-go / posthog-rs). ``Retry-After`` is itself clamped to + :data:`_MAX_BACKOFF_SECONDS`, so both sides share one ceiling and a + hostile/buggy header can't park the consumer thread. + """ + configured = min(2**attempt_index, _MAX_BACKOFF_SECONDS) + clamped_retry_after = ( + min(retry_after, _MAX_BACKOFF_SECONDS) if retry_after and retry_after > 0 else 0 + ) + time.sleep(max(configured, clamped_retry_after)) + + +def _log_result_summary( + request_id: str, attempt: int, results: dict[str, _V1EventResult] +) -> None: + tally = {_RESULT_OK: 0, _RESULT_WARNING: 0, _RESULT_DROP: 0, _RESULT_RETRY: 0} + other = 0 + for r in results.values(): + if r.result in tally: + tally[r.result] += 1 + else: + other += 1 + log.debug( + "capture v1 response request_id=%s attempt=%s events=%d ok=%d warning=%d drop=%d retry=%d other=%d", + request_id, + attempt, + len(results), + tally[_RESULT_OK], + tally[_RESULT_WARNING], + tally[_RESULT_DROP], + tally[_RESULT_RETRY], + other, + ) + + +def _send_v1_batch( + api_key: str, + host: Optional[str], + batch: list[dict], + *, + compression: CaptureCompression = CaptureCompression.NONE, + timeout: int = 15, + max_retries: int = 3, + historical_migration: bool = False, + session: Optional["requests.Session"] = None, +) -> None: + """Deliver ``batch`` to the v1 endpoint with partial retry. + + The v1 sibling of ``Consumer._send``: it loops up to ``max_retries + 1`` + attempts, but unlike v0 it shrinks the batch to only the events the server + tagged ``retry`` after each 2xx. ``ok``/``warning``/absent events succeed. + + A server-chosen ``drop`` is a terminal per-event rejection. Drops are + accumulated across attempts and surfaced via :class:`CaptureV1Error` even + when the request itself was a 2xx (a success status is not full delivery) + and even when a later attempt clears the outstanding retries — matching + posthog-go (per-event failure callback) and posthog-rs (``on_error`` on a + 2xx with undelivered verdicts). Raises :class:`CaptureV1Error` on any drop, + batch-level terminal failure, or retry exhaustion — carrying the accumulated + ``drops`` and any exhausted uuids — so the caller's ``on_error`` fires + unchanged. A transport failure re-raises the underlying exception (drops + collected on an earlier attempt are still tallied in the DEBUG summary). + ``request_id`` and the batch ``created_at`` are stable across attempts; + ``PostHog-Attempt`` increments. + """ + request_id = str(uuid4()) + # Hoisted once so the batch envelope is byte-identical across retry attempts + # (only the events list shrinks and the attempt header increments). + created_at = datetime.now(timezone.utc).isoformat() + pending_events = [_to_v1_event(m) for m in batch] + pending_uuids = [e["uuid"] for e in pending_events] + last_exc: Optional[Exception] = None + # (uuid, details) for every event the server dropped, across all attempts. + # Accumulated (not per-attempt) so a drop seen early is not lost when a + # later attempt succeeds or clears the outstanding retries. + all_drops: list[tuple[str, Optional[str]]] = [] + + for attempt_index in range(max_retries + 1): + attempt = attempt_index + 1 + last_attempt = attempt_index == max_retries + body = _build_v1_batch_body( + pending_events, historical_migration, created_at=created_at + ) + + try: + res = _post_v1( + api_key, + host, + body, + attempt=attempt, + request_id=request_id, + compression=compression, + timeout=timeout, + session=session, + ) + except Exception as e: + # Transport-level failure (connection/timeout): retry like v0 does. + last_exc = e + if last_attempt: + raise + _backoff(attempt_index, None) + continue + + parsed = _parse_v1_response(res) + + if parsed.is_success: + if parsed.malformed: + raise CaptureV1Error( + parsed.status_code, + "capture v1 returned a success status with an unparseable body", + request_id=request_id, + attempts=attempt, + drops=all_drops, + ) + results = parsed.results or {} + _log_result_summary(request_id, attempt, results) + + retry_events: list[dict] = [] + retry_uuids: list[str] = [] + for event, uid in zip(pending_events, pending_uuids): + directive = results.get(uid) + if directive is None: + # Absent from the map: treated as accepted (matches posthog-rs). + continue + if directive.result == _RESULT_RETRY: + retry_events.append(event) + retry_uuids.append(uid) + elif directive.result == _RESULT_DROP: + # Terminal per-event rejection; keep it so it is surfaced + # even when the rest of the batch succeeds (see below). + all_drops.append((uid, directive.details)) + # ok / warning / unrecognized -> terminal success. + + if not retry_uuids: + # Nothing left to resend. If the server dropped any events, + # surface them via on_error even though the request was a 2xx — + # a success status does not mean every event was delivered. + if all_drops: + raise CaptureV1Error( + parsed.status_code, + f"{len(all_drops)} event(s) dropped by the server", + request_id=request_id, + attempts=attempt, + drops=all_drops, + ) + return + if last_attempt: + raise CaptureV1Error( + parsed.status_code, + f"{len(retry_uuids)} event(s) still pending retry after {attempt} attempt(s)", + request_id=request_id, + attempts=attempt, + retry_exhausted=retry_uuids, + drops=all_drops, + ) + pending_events, pending_uuids = retry_events, retry_uuids + _backoff(attempt_index, parsed.retry_after) + continue + + # Non-2xx. Retryable transient statuses back off; everything else + # (400/401/402/413/415/429/...) is terminal. Any drops collected from a + # prior 2xx attempt ride along so on_error still sees them. + v1_error = CaptureV1Error( + parsed.status_code, + parsed.error_message, + retry_after=parsed.retry_after, + request_id=request_id, + attempts=attempt, + drops=all_drops, + ) + if parsed.status_code in _RETRYABLE_STATUSES: + last_exc = v1_error + if last_attempt: + raise v1_error + _backoff(attempt_index, parsed.retry_after) + continue + raise v1_error + + # Unreachable in practice (every branch returns or continues), but keeps the + # function total if max_retries is somehow negative. + if last_exc: + raise last_exc diff --git a/posthog/client.py b/posthog/client.py index 99ef4801..b5136b72 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -16,6 +16,12 @@ from posthog._async_utils import _BackgroundEventLoopRunner from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs +from posthog.capture_compression import ( + CaptureCompression, + _resolve_capture_compression, +) +from posthog.capture_mode import CaptureMode, _resolve_capture_mode +from posthog.capture_v1 import _send_v1_batch from posthog.consumer import Consumer from posthog.contexts import ( _get_current_context, @@ -260,6 +266,8 @@ def __init__( exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, + capture_mode: Optional[Union[CaptureMode, str]] = None, + capture_compression: Optional[Union[CaptureCompression, str]] = None, _dedicated_ai_endpoint=False, ): """ @@ -343,6 +351,16 @@ def __init__( interval for each exception type's bucket. exception_autocapture_refill_interval_seconds: Seconds between token refills for autocaptured exception rate limiting. + capture_mode: Capture wire protocol to use. Defaults to + ``CaptureMode.V0`` (legacy ``/batch/``). Set ``CaptureMode.V1`` + (or pass the string ``"v1"``) to opt into + ``/i/v1/analytics/events``. When omitted, the + ``POSTHOG_CAPTURE_MODE`` env var is consulted, then ``V0``. + capture_compression: Request-body compression for capture-v1 uploads + (ignored in V0, which uses ``gzip``). ``CaptureCompression.GZIP`` + or ``DEFLATE`` (or the strings ``"gzip"``/``"deflate"``). When + omitted, the ``POSTHOG_CAPTURE_COMPRESSION`` env var is consulted, + then the legacy ``gzip`` flag, then no compression. Examples: ```python @@ -370,6 +388,7 @@ def __init__( self._duplicate_client_registry_key: Optional[tuple[str, str]] = None self.gzip = gzip self.timeout = timeout + self.max_retries = max_retries self._feature_flags: Optional[list[Any]] = ( None # private variable to store flags ) @@ -398,6 +417,15 @@ def __init__( self.disable_geoip = disable_geoip self.is_server = is_server self.historical_migration = historical_migration + # Selects the capture wire protocol (V0 legacy `/batch/` vs V1 + # `/i/v1/analytics/events`). Resolved here so the env-var fallback is + # applied once; V0 is the default and keeps upgrades transparent. + self.capture_mode = _resolve_capture_mode(capture_mode) + # v1-only request compression; falls back to the legacy `gzip` flag when + # neither the kwarg nor POSTHOG_CAPTURE_COMPRESSION is set. + self.capture_compression = _resolve_capture_compression( + capture_compression, gzip_fallback=gzip + ) # Internal, not ready for use: routes `$ai_*` events to a dedicated # capture-ai endpoint while the backend route + ingress roll out. self._dedicated_ai_endpoint = _dedicated_ai_endpoint @@ -503,6 +531,8 @@ def __init__( timeout=timeout, historical_migration=historical_migration, dedicated_ai_endpoint=self._dedicated_ai_endpoint, + capture_mode=self.capture_mode, + capture_compression=self.capture_compression, ) self.consumers.append(consumer) @@ -1525,6 +1555,8 @@ def _reinit_after_fork(self): timeout=old.timeout, historical_migration=old.historical_migration, dedicated_ai_endpoint=old.dedicated_ai_endpoint, + capture_mode=old.capture_mode, + capture_compression=old.capture_compression, ) new_consumers.append(consumer) @@ -1624,11 +1656,24 @@ def _enqueue(self, msg, disable_geoip): if self.sync_mode: self.log.debug("enqueued with blocking %s.", msg["event"]) - path = ( - AI_EVENTS_ENDPOINT - if self._dedicated_ai_endpoint and is_ai_event(msg.get("event")) - else EVENTS_ENDPOINT + is_dedicated_ai = self._dedicated_ai_endpoint and is_ai_event( + msg.get("event") ) + # Analytics events follow `capture_mode`; the dedicated AI endpoint + # has no v1 form and always uses the legacy submitter. + if not is_dedicated_ai and self.capture_mode == CaptureMode.V1: + _send_v1_batch( + self.api_key, + self.host, + [msg], + compression=self.capture_compression, + timeout=self.timeout, + max_retries=self.max_retries, + historical_migration=self.historical_migration, + ) + return sent_uuid + + path = AI_EVENTS_ENDPOINT if is_dedicated_ai else EVENTS_ENDPOINT batch_post( self.api_key, self.host, diff --git a/posthog/consumer.py b/posthog/consumer.py index 567292e8..0abd24ca 100644 --- a/posthog/consumer.py +++ b/posthog/consumer.py @@ -5,6 +5,9 @@ from threading import Thread from posthog._logging import _configure_posthog_logging +from posthog.capture_compression import CaptureCompression +from posthog.capture_mode import CaptureMode +from posthog.capture_v1 import _send_v1_batch from posthog.request import ( AI_EVENTS_ENDPOINT, EVENTS_ENDPOINT, @@ -50,6 +53,8 @@ def __init__( timeout=15, historical_migration=False, dedicated_ai_endpoint=False, + capture_mode=CaptureMode.V0, + capture_compression=CaptureCompression.NONE, ): """Create a consumer thread.""" Thread.__init__(self) @@ -63,6 +68,8 @@ def __init__( self.queue = queue self.gzip = gzip self.dedicated_ai_endpoint = dedicated_ai_endpoint + self.capture_mode = capture_mode + self.capture_compression = capture_compression # It's important to set running in the constructor: if we are asked to # pause immediately after construction, we might set running to True in # run() *after* we set it to False in pause... and keep running @@ -151,9 +158,13 @@ def request(self, batch): invokes `on_error`); a second is logged here so it isn't silently lost. The batch was already dequeued in `upload()`, so unsent events are dropped after retries, same as the single-endpoint path. + + The analytics destination follows `capture_mode` (v1 -> partial-retry + submitter); the dedicated AI endpoint has no v1 form and always uses the + legacy submitter. """ if not self.dedicated_ai_endpoint: - self._send(batch, EVENTS_ENDPOINT) + self._send_analytics(batch) return ai_events: list[Any] = [] @@ -163,23 +174,42 @@ def request(self, batch): target.append(item) first_exc = None - for events, path in ( - (analytics_events, EVENTS_ENDPOINT), - (ai_events, AI_EVENTS_ENDPOINT), + for events, label, sender in ( + (analytics_events, "analytics", self._send_analytics), + (ai_events, "ai", self._send_ai), ): if not events: continue try: - self._send(events, path) + sender(events) except Exception as e: if first_exc is None: first_exc = e else: - self.log.error("error uploading to %s: %s", path, e) + self.log.error("error uploading to %s: %s", label, e) if first_exc is not None: raise first_exc + def _send_analytics(self, batch): + """Submit analytics events via the wire protocol selected by `capture_mode`.""" + if self.capture_mode == CaptureMode.V1: + _send_v1_batch( + self.api_key, + self.host, + batch, + compression=self.capture_compression, + timeout=self.timeout, + max_retries=self.retries, + historical_migration=self.historical_migration, + ) + return + self._send(batch, EVENTS_ENDPOINT) + + def _send_ai(self, batch): + """Submit `$ai_*` events to the dedicated legacy AI endpoint (no v1 form).""" + self._send(batch, AI_EVENTS_ENDPOINT) + def _send(self, batch, path): """Attempt to upload a single batch to `path`, retrying before raising an error""" diff --git a/posthog/test/test_capture_compression.py b/posthog/test/test_capture_compression.py new file mode 100644 index 00000000..a4ce0154 --- /dev/null +++ b/posthog/test/test_capture_compression.py @@ -0,0 +1,168 @@ +import os +import unittest +from unittest import mock + +from parameterized import parameterized + +from posthog.capture_compression import ( + CAPTURE_COMPRESSION_ENV_VAR, + CaptureCompression, + _resolve_capture_compression, +) +from posthog.client import Client +from posthog.consumer import Consumer +from posthog.test.logging_helpers import capture_message_only_logs +from posthog.test.test_utils import TEST_API_KEY + + +class TestResolveCaptureCompression(unittest.TestCase): + def setUp(self) -> None: + patcher = mock.patch.dict(os.environ, {}, clear=False) + patcher.start() + self.addCleanup(patcher.stop) + os.environ.pop(CAPTURE_COMPRESSION_ENV_VAR, None) + + def test_defaults_to_none_with_no_kwarg_env_or_gzip(self) -> None: + self.assertIs(_resolve_capture_compression(None), CaptureCompression.NONE) + + def test_gzip_fallback_used_when_nothing_else_set(self) -> None: + self.assertIs( + _resolve_capture_compression(None, gzip_fallback=True), + CaptureCompression.GZIP, + ) + + @parameterized.expand( + [ + ("enum_gzip", CaptureCompression.GZIP, CaptureCompression.GZIP), + ("enum_deflate", CaptureCompression.DEFLATE, CaptureCompression.DEFLATE), + ("enum_none", CaptureCompression.NONE, CaptureCompression.NONE), + ("enum_zstd", CaptureCompression.ZSTD, CaptureCompression.ZSTD), + ("str_gzip", "gzip", CaptureCompression.GZIP), + ("str_deflate", "deflate", CaptureCompression.DEFLATE), + ("str_none", "none", CaptureCompression.NONE), + ("str_zstd", "zstd", CaptureCompression.ZSTD), + ("str_identity_alias", "identity", CaptureCompression.NONE), + ("str_upper_and_padded", " GZIP ", CaptureCompression.GZIP), + ] + ) + def test_explicit_kwarg_takes_precedence_and_coerces( + self, _name, kwarg, expected + ) -> None: + # Env names a different value and gzip_fallback is on, so each row proves + # the explicit kwarg wins over both lower-precedence sources. + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: "deflate"}): + self.assertIs( + _resolve_capture_compression(kwarg, gzip_fallback=True), expected + ) + + def test_invalid_kwarg_raises_even_with_valid_env(self) -> None: + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: "gzip"}): + with self.assertRaises(ValueError): + _resolve_capture_compression("bogus") + + @parameterized.expand([("bad_str", "bogus"), ("wrong_type", 1)]) + def test_invalid_explicit_kwarg_raises(self, _name, value) -> None: + with self.assertRaises(ValueError): + _resolve_capture_compression(value) + + @parameterized.expand( + [ + ("gzip", "gzip", CaptureCompression.GZIP), + ("deflate", "deflate", CaptureCompression.DEFLATE), + ("none", "none", CaptureCompression.NONE), + ("zstd", "zstd", CaptureCompression.ZSTD), + ("identity", "identity", CaptureCompression.NONE), + ("uppercase", "GZIP", CaptureCompression.GZIP), + ("padded", " deflate ", CaptureCompression.DEFLATE), + ] + ) + def test_env_var_resolution(self, _name, env_value, expected) -> None: + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: env_value}): + self.assertIs(_resolve_capture_compression(None), expected) + + def test_env_var_takes_precedence_over_gzip_fallback(self) -> None: + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: "deflate"}): + self.assertIs( + _resolve_capture_compression(None, gzip_fallback=True), + CaptureCompression.DEFLATE, + ) + + @parameterized.expand([("empty", ""), ("whitespace", " ")]) + def test_blank_env_var_falls_through_to_fallback(self, _name, env_value) -> None: + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: env_value}): + self.assertIs(_resolve_capture_compression(None), CaptureCompression.NONE) + self.assertIs( + _resolve_capture_compression(None, gzip_fallback=True), + CaptureCompression.GZIP, + ) + + def test_unrecognized_env_var_warns_and_uses_fallback(self) -> None: + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: "bogus"}): + with capture_message_only_logs() as stream: + self.assertIs( + _resolve_capture_compression(None, gzip_fallback=True), + CaptureCompression.GZIP, + ) + self.assertIn("bogus", stream.getvalue()) + + @parameterized.expand([("enum", CaptureCompression.ZSTD), ("str", "zstd")]) + def test_explicit_zstd_without_package_raises(self, _name, kwarg) -> None: + with mock.patch("posthog.capture_compression._zstandard", None): + with self.assertRaises(ValueError) as ctx: + _resolve_capture_compression(kwarg) + self.assertIn("posthog[zstd]", str(ctx.exception)) + + def test_env_zstd_without_package_warns_and_uses_fallback(self) -> None: + with mock.patch("posthog.capture_compression._zstandard", None): + with mock.patch.dict(os.environ, {CAPTURE_COMPRESSION_ENV_VAR: "zstd"}): + with capture_message_only_logs() as stream: + self.assertIs( + _resolve_capture_compression(None, gzip_fallback=True), + CaptureCompression.GZIP, + ) + self.assertIn("posthog[zstd]", stream.getvalue()) + + +class TestCaptureCompressionPlumbing(unittest.TestCase): + def setUp(self) -> None: + patcher = mock.patch.dict(os.environ, {}, clear=False) + patcher.start() + self.addCleanup(patcher.stop) + os.environ.pop(CAPTURE_COMPRESSION_ENV_VAR, None) + + def test_client_defaults_to_none(self) -> None: + client = Client(TEST_API_KEY, sync_mode=True) + self.assertIs(client.capture_compression, CaptureCompression.NONE) + + def test_client_gzip_flag_falls_back_to_gzip(self) -> None: + client = Client(TEST_API_KEY, sync_mode=True, gzip=True) + self.assertIs(client.capture_compression, CaptureCompression.GZIP) + + @parameterized.expand( + [ + ("enum_deflate", CaptureCompression.DEFLATE, CaptureCompression.DEFLATE), + ("str_gzip", "gzip", CaptureCompression.GZIP), + ("str_none", "none", CaptureCompression.NONE), + ] + ) + def test_client_kwarg_overrides_gzip_flag(self, _name, kwarg, expected) -> None: + # Even with the legacy gzip flag on, the explicit kwarg wins. + client = Client( + TEST_API_KEY, sync_mode=True, gzip=True, capture_compression=kwarg + ) + self.assertIs(client.capture_compression, expected) + + def test_client_propagates_to_consumers(self) -> None: + client = Client( + TEST_API_KEY, + capture_compression=CaptureCompression.DEFLATE, + send=False, + thread=2, + ) + self.assertEqual(len(client.consumers), 2) + for consumer in client.consumers: + self.assertIs(consumer.capture_compression, CaptureCompression.DEFLATE) + + def test_consumer_defaults_to_none(self) -> None: + consumer = Consumer(None, TEST_API_KEY) + self.assertIs(consumer.capture_compression, CaptureCompression.NONE) diff --git a/posthog/test/test_capture_mode.py b/posthog/test/test_capture_mode.py new file mode 100644 index 00000000..473e1a88 --- /dev/null +++ b/posthog/test/test_capture_mode.py @@ -0,0 +1,109 @@ +import os +import unittest +from unittest import mock + +from parameterized import parameterized + +from posthog.capture_mode import ( + CAPTURE_MODE_ENV_VAR, + CaptureMode, + _resolve_capture_mode, +) +from posthog.client import Client +from posthog.consumer import Consumer +from posthog.test.logging_helpers import capture_message_only_logs +from posthog.test.test_utils import TEST_API_KEY + + +class TestResolveCaptureMode(unittest.TestCase): + def test_defaults_to_v0_with_no_kwarg_and_no_env(self) -> None: + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop(CAPTURE_MODE_ENV_VAR, None) + self.assertIs(_resolve_capture_mode(None), CaptureMode.V0) + + @parameterized.expand( + [ + # (name, kwarg, expected, opposite_env): the env always names the + # mode the kwarg must override, so every row proves the kwarg wins. + ("enum_v0", CaptureMode.V0, CaptureMode.V0, "v1"), + ("enum_v1", CaptureMode.V1, CaptureMode.V1, "v0"), + ("str_v0", "v0", CaptureMode.V0, "v1"), + ("str_v1", "v1", CaptureMode.V1, "v0"), + ("str_legacy_alias", "legacy", CaptureMode.V0, "v1"), + ("str_analytics_v1_alias", "analytics_v1", CaptureMode.V1, "v0"), + ("str_upper_and_padded", " V1 ", CaptureMode.V1, "v0"), + ] + ) + def test_explicit_kwarg_takes_precedence_and_coerces( + self, _name, kwarg, expected, opposite_env + ) -> None: + with mock.patch.dict(os.environ, {CAPTURE_MODE_ENV_VAR: opposite_env}): + self.assertIs(_resolve_capture_mode(kwarg), expected) + + def test_invalid_kwarg_raises_even_with_valid_env(self) -> None: + # The kwarg path is consulted before the env, so an invalid kwarg raises + # rather than silently falling back to a valid env value. + with mock.patch.dict(os.environ, {CAPTURE_MODE_ENV_VAR: "v1"}): + with self.assertRaises(ValueError): + _resolve_capture_mode("bogus") + + @parameterized.expand( + [ + ("v0", "v0", CaptureMode.V0), + ("legacy", "legacy", CaptureMode.V0), + ("v1", "v1", CaptureMode.V1), + ("analytics_v1", "analytics_v1", CaptureMode.V1), + ("uppercase", "V1", CaptureMode.V1), + ("padded", " v1 ", CaptureMode.V1), + ] + ) + def test_env_var_resolution(self, _name, env_value, expected) -> None: + with mock.patch.dict(os.environ, {CAPTURE_MODE_ENV_VAR: env_value}): + self.assertIs(_resolve_capture_mode(None), expected) + + @parameterized.expand([("empty", ""), ("whitespace", " ")]) + def test_blank_env_var_defaults_to_v0(self, _name, env_value) -> None: + with mock.patch.dict(os.environ, {CAPTURE_MODE_ENV_VAR: env_value}): + self.assertIs(_resolve_capture_mode(None), CaptureMode.V0) + + def test_unrecognized_env_var_warns_and_defaults_to_v0(self) -> None: + with mock.patch.dict(os.environ, {CAPTURE_MODE_ENV_VAR: "bogus"}): + with capture_message_only_logs() as stream: + self.assertIs(_resolve_capture_mode(None), CaptureMode.V0) + self.assertIn("bogus", stream.getvalue()) + + @parameterized.expand([("bad_str", "bogus"), ("wrong_type", 1)]) + def test_invalid_explicit_kwarg_raises(self, _name, value) -> None: + with self.assertRaises(ValueError): + _resolve_capture_mode(value) + + +class TestCaptureModePlumbing(unittest.TestCase): + def test_client_resolves_and_stores_default_v0(self) -> None: + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop(CAPTURE_MODE_ENV_VAR, None) + client = Client(TEST_API_KEY, sync_mode=True) + self.assertIs(client.capture_mode, CaptureMode.V0) + + @parameterized.expand( + [ + ("enum_v1", CaptureMode.V1, CaptureMode.V1), + ("str_v1", "v1", CaptureMode.V1), + ("enum_v0", CaptureMode.V0, CaptureMode.V0), + ] + ) + def test_client_kwarg_sets_mode(self, _name, kwarg, expected) -> None: + client = Client(TEST_API_KEY, sync_mode=True, capture_mode=kwarg) + self.assertIs(client.capture_mode, expected) + + def test_client_propagates_mode_to_consumers(self) -> None: + # Async (non-sync) client builds Consumer threads; assert each carries + # the resolved mode. + client = Client(TEST_API_KEY, capture_mode=CaptureMode.V1, send=False, thread=2) + self.assertEqual(len(client.consumers), 2) + for consumer in client.consumers: + self.assertIs(consumer.capture_mode, CaptureMode.V1) + + def test_consumer_defaults_to_v0(self) -> None: + consumer = Consumer(None, TEST_API_KEY) + self.assertIs(consumer.capture_mode, CaptureMode.V0) diff --git a/posthog/test/test_capture_v1.py b/posthog/test/test_capture_v1.py new file mode 100644 index 00000000..a368ae5d --- /dev/null +++ b/posthog/test/test_capture_v1.py @@ -0,0 +1,711 @@ +import json +import unittest +import zlib +from datetime import datetime, timezone +from unittest import mock + +import zstandard + +from parameterized import parameterized + +from posthog.capture_compression import CaptureCompression +from posthog.capture_v1 import ( + _CAPTURE_V1_PATH, + _HEADER_ATTEMPT, + _HEADER_REQUEST_ID, + _HEADER_SDK_INFO, + _MAX_BACKOFF_SECONDS, + CaptureV1Error, + _build_v1_batch_body, + _parse_v1_response, + _post_v1, + _send_v1_batch, + _to_v1_event, + _backoff, + _coerce_bool, + _coerce_str, +) + + +class _FakeResponse: + """Minimal stand-in for ``requests.Response`` for transport tests.""" + + def __init__( + self, status_code, *, json_body=None, headers=None, text="", raise_json=False + ): + self.status_code = status_code + self.headers = headers or {} + self.text = text + self._json_body = json_body + self._raise_json = raise_json + + def json(self): + if self._raise_json: + raise ValueError("no json") + return self._json_body + + +class _RecordingSession: + """Captures the args of a single ``.post`` and returns a canned response.""" + + def __init__(self, response): + self._response = response + self.calls = [] + + def post(self, url, data=None, headers=None, timeout=None): + self.calls.append( + {"url": url, "data": data, "headers": headers, "timeout": timeout} + ) + return self._response + + +class _PostV1Stub: + """Drop-in for ``_post_v1`` that records calls and replays canned outcomes. + + Each item in ``outcomes`` is either a ``_FakeResponse`` to return or an + ``Exception`` instance to raise (simulating a transport failure). + """ + + def __init__(self, outcomes): + self._outcomes = list(outcomes) + self.calls = [] + + def __call__( + self, + api_key, + host, + batch_body, + *, + attempt, + request_id, + compression=CaptureCompression.NONE, + timeout=15, + session=None, + ): + self.calls.append( + { + "attempt": attempt, + "request_id": request_id, + "compression": compression, + "created_at": batch_body["created_at"], + "uuids": [e["uuid"] for e in batch_body["batch"]], + } + ) + outcome = self._outcomes[len(self.calls) - 1] + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def _msg(uuid, event="e", **overrides): + msg = { + "event": event, + "uuid": uuid, + "distinct_id": "user-1", + "timestamp": "2026-06-27T12:00:00+00:00", + "type": "capture", + "properties": {}, + } + msg.update(overrides) + return msg + + +def _results_response(directives, headers=None): + """200 response whose ``results`` map tags each uuid. + + ``directives`` maps uuid -> ``"ok"`` (or any result string) or a + ``(result, details)`` tuple. + """ + results = {} + for uid, spec in directives.items(): + result, details = spec if isinstance(spec, tuple) else (spec, None) + results[uid] = {"result": result, "details": details} + return _FakeResponse(200, json_body={"results": results}, headers=headers) + + +def _legacy_msg(event="my_event", properties=None, **overrides) -> dict: + """Minimal legacy-shaped message as it looks coming off the queue.""" + msg = { + "event": event, + "uuid": "0190000000007000800000000000000a", + "distinct_id": "user-1", + "timestamp": "2026-06-27T12:00:00+00:00", + "type": "capture", + "properties": {"$lib": "posthog-python", "$lib_version": "9.9.9"}, + } + if properties is not None: + msg["properties"] = properties + msg.update(overrides) + return msg + + +class TestCoercion(unittest.TestCase): + @parameterized.expand( + [ + ("bool_true", True, True), + ("bool_false", False, False), + ("str_true", "true", True), + ("str_true_upper", "TRUE", True), + ("str_true_padded", " true ", True), + ("str_one", "1", True), + ("str_false", "false", False), + ("str_zero", "0", False), + ("int_nonzero", 5, True), + ("int_zero", 0, False), + ("float_nonzero", 1.5, True), + ("float_zero", 0.0, False), + ("neg_int", -1, True), + ("str_yes_uncoercible", "yes", None), + ("str_empty_uncoercible", "", None), + ("none_uncoercible", None, None), + ("dict_uncoercible", {"a": 1}, None), + ] + ) + def test_coerce_bool(self, _name, value, expected) -> None: + self.assertIs(_coerce_bool(value), expected) + + @parameterized.expand( + [ + ("str", "tour-1", "tour-1"), + ("empty_str", "", ""), + ("int", 123, None), + ("bool", True, None), + ("none", None, None), + ] + ) + def test_coerce_str(self, _name, value, expected) -> None: + self.assertEqual(_coerce_str(value), expected) + + +class TestToV1Event(unittest.TestCase): + def test_required_fields_preserved(self) -> None: + event = _to_v1_event(_legacy_msg(event="signed_up")) + self.assertEqual(event["event"], "signed_up") + self.assertEqual(event["uuid"], "0190000000007000800000000000000a") + self.assertEqual(event["distinct_id"], "user-1") + self.assertEqual(event["timestamp"], "2026-06-27T12:00:00+00:00") + + def test_strips_lib_and_lib_version(self) -> None: + event = _to_v1_event(_legacy_msg()) + self.assertNotIn("$lib", event["properties"]) + self.assertNotIn("$lib_version", event["properties"]) + + def test_options_empty_dict_when_no_sentinels(self) -> None: + event = _to_v1_event(_legacy_msg(properties={"plain": "value"})) + self.assertEqual(event["options"], {}) + self.assertEqual(event["properties"], {"plain": "value"}) + + def test_does_not_leak_non_wire_top_level_keys(self) -> None: + event = _to_v1_event(_legacy_msg()) + # `type` is legacy-only; the v1 event carries only documented fields. + self.assertEqual( + set(event), + {"event", "uuid", "distinct_id", "timestamp", "options", "properties"}, + ) + + def test_does_not_mutate_input(self) -> None: + msg = _legacy_msg( + properties={"$cookieless_mode": True, "$session_id": "s-1"}, + **{"$set": {"name": "Max"}}, + ) + original_properties = dict(msg["properties"]) + _to_v1_event(msg) + self.assertEqual(msg["properties"], original_properties) + self.assertIn("$set", msg) # top-level $set untouched on the original + + @parameterized.expand( + [ + ("cookieless_mode", "$cookieless_mode", "cookieless_mode", True, True), + ( + "ignore_sent_at_rename", + "$ignore_sent_at", + "disable_skew_correction", + "true", + True, + ), + ( + "process_person_profile", + "$process_person_profile", + "process_person_profile", + "false", + False, + ), + ( + "product_tour_id", + "$product_tour_id", + "product_tour_id", + "tour-7", + "tour-7", + ), + ] + ) + def test_option_sentinels_lifted_renamed_and_coerced( + self, _name, prop_key, wire_key, raw, expected + ) -> None: + event = _to_v1_event(_legacy_msg(properties={prop_key: raw})) + self.assertEqual(event["options"], {wire_key: expected}) + self.assertNotIn(prop_key, event["properties"]) + + @parameterized.expand( + [ + ("bad_bool", "$cookieless_mode", "maybe"), + ("bad_tour_id_int", "$product_tour_id", 123), + ] + ) + def test_option_sentinel_removed_but_omitted_on_bad_coercion( + self, _name, prop_key, raw + ) -> None: + event = _to_v1_event(_legacy_msg(properties={prop_key: raw})) + # Removed from properties (sentinels must never reach v1 props) but not + # emitted as an option, so a wrong type cannot 400 the whole batch. + self.assertNotIn(prop_key, event["properties"]) + self.assertEqual(event["options"], {}) + + @parameterized.expand( + [ + ("session_id", "$session_id", "session_id", "s-123"), + ("window_id", "$window_id", "window_id", "w-456"), + ] + ) + def test_top_level_string_sentinels(self, _name, prop_key, field_name, raw) -> None: + event = _to_v1_event(_legacy_msg(properties={prop_key: raw})) + self.assertEqual(event[field_name], raw) + self.assertNotIn(prop_key, event["properties"]) + + def test_top_level_sentinel_omitted_but_removed_when_not_string(self) -> None: + event = _to_v1_event(_legacy_msg(properties={"$session_id": 42})) + self.assertNotIn("session_id", event) + self.assertNotIn("$session_id", event["properties"]) + + def test_all_sentinels_together(self) -> None: + event = _to_v1_event( + _legacy_msg( + properties={ + "$cookieless_mode": True, + "$ignore_sent_at": "1", + "$product_tour_id": "tour-x", + "$process_person_profile": 0, + "$session_id": "s-1", + "$window_id": "w-1", + "$geoip_disable": True, + "custom": "keep", + } + ) + ) + self.assertEqual( + event["options"], + { + "cookieless_mode": True, + "disable_skew_correction": True, + "product_tour_id": "tour-x", + "process_person_profile": False, + }, + ) + self.assertEqual(event["session_id"], "s-1") + self.assertEqual(event["window_id"], "w-1") + # Non-sentinel props (including $geoip_disable) are left intact. + self.assertEqual( + event["properties"], {"$geoip_disable": True, "custom": "keep"} + ) + + @parameterized.expand([("set", "$set"), ("set_once", "$set_once")]) + def test_top_level_set_relocated_into_properties(self, _name, key) -> None: + msg = _legacy_msg(properties={}, **{key: {"email": "a@b.com"}}) + event = _to_v1_event(msg) + self.assertEqual(event["properties"][key], {"email": "a@b.com"}) + self.assertNotIn(key, event) # not a top-level v1 field + + def test_top_level_set_merges_with_existing_properties_set(self) -> None: + # properties wins on key collision. + msg = _legacy_msg( + properties={"$set": {"a": "from_props", "b": "props_only"}}, + **{"$set": {"a": "from_top", "c": "top_only"}}, + ) + event = _to_v1_event(msg) + self.assertEqual( + event["properties"]["$set"], + {"a": "from_props", "b": "props_only", "c": "top_only"}, + ) + + def test_groups_left_in_properties(self) -> None: + event = _to_v1_event(_legacy_msg(properties={"$groups": {"company": "ph"}})) + self.assertEqual(event["properties"]["$groups"], {"company": "ph"}) + + def test_timestamp_naive_datetime_made_tz_aware(self) -> None: + event = _to_v1_event(_legacy_msg(timestamp=datetime(2026, 6, 27, 12, 0, 0))) + parsed = datetime.fromisoformat(event["timestamp"]) + self.assertIsNotNone(parsed.tzinfo) + + def test_timestamp_none_defaults_to_utc_now(self) -> None: + event = _to_v1_event(_legacy_msg(timestamp=None)) + parsed = datetime.fromisoformat(event["timestamp"]) + self.assertEqual(parsed.tzinfo, timezone.utc) + + +class TestBuildV1BatchBody(unittest.TestCase): + def test_envelope_shape_and_no_legacy_fields(self) -> None: + events = [{"event": "e"}] + body = _build_v1_batch_body(events) + self.assertEqual(body["batch"], events) + self.assertNotIn("api_key", body) + self.assertNotIn("sent_at", body) + + def test_created_at_is_tz_aware_rfc3339(self) -> None: + body = _build_v1_batch_body([]) + parsed = datetime.fromisoformat(body["created_at"]) + self.assertIsNotNone(parsed.tzinfo) + + def test_created_at_passthrough_used_verbatim(self) -> None: + # _send_v1_batch hoists created_at and passes it in so it stays stable + # across retry attempts. + body = _build_v1_batch_body([], created_at="2026-06-27T12:00:00+00:00") + self.assertEqual(body["created_at"], "2026-06-27T12:00:00+00:00") + + def test_historical_migration_omitted_when_false(self) -> None: + self.assertNotIn("historical_migration", _build_v1_batch_body([])) + + def test_historical_migration_present_when_true(self) -> None: + body = _build_v1_batch_body([], historical_migration=True) + self.assertIs(body["historical_migration"], True) + + +class TestPostV1(unittest.TestCase): + def _post(self, response, **kwargs): + session = _RecordingSession(response) + body = _build_v1_batch_body([_to_v1_event(_msg("u-1"))]) + _post_v1( + "phc_key", + "https://app.posthog.com/", + body, + attempt=2, + request_id="req-123", + session=session, + **kwargs, + ) + return session.calls[0] + + def test_url_uses_v1_path_and_trims_host(self) -> None: + call = self._post(_results_response({})) + self.assertEqual(call["url"], "https://app.posthog.com" + _CAPTURE_V1_PATH) + + def test_required_headers_present(self) -> None: + headers = self._post(_results_response({}))["headers"] + self.assertEqual(headers["Authorization"], "Bearer phc_key") + self.assertEqual(headers[_HEADER_ATTEMPT], "2") + self.assertEqual(headers[_HEADER_REQUEST_ID], "req-123") + self.assertTrue(headers[_HEADER_SDK_INFO].startswith("posthog-python/")) + self.assertEqual(headers["Content-Type"], "application/json") + + def test_no_api_key_in_body(self) -> None: + # v1 authenticates via the Bearer header; the key must not leak into the body. + data = self._post(_results_response({}))["data"] + self.assertNotIn("phc_key", data) + self.assertNotIn("api_key", json.loads(data)) + + def test_uncompressed_body_is_json_str_without_encoding_header(self) -> None: + call = self._post(_results_response({}), compression=CaptureCompression.NONE) + self.assertIsInstance(call["data"], str) + self.assertNotIn("Content-Encoding", call["headers"]) + + def test_gzip_sets_encoding_header_and_compresses_body(self) -> None: + call = self._post(_results_response({}), compression=CaptureCompression.GZIP) + self.assertEqual(call["headers"]["Content-Encoding"], "gzip") + self.assertIsInstance(call["data"], bytes) + self.assertEqual(call["data"][:2], b"\x1f\x8b") # gzip magic + + def test_deflate_sets_encoding_header_and_zlib_wraps_body(self) -> None: + # Must be zlib-wrapped (RFC 1950, 0x78 prefix), matching posthog-go / + # posthog-rs, so the server routes Content-Encoding: deflate to its zlib + # decoder rather than treating it as raw deflate. + call = self._post(_results_response({}), compression=CaptureCompression.DEFLATE) + self.assertEqual(call["headers"]["Content-Encoding"], "deflate") + self.assertIsInstance(call["data"], bytes) + self.assertEqual(call["data"][0], 0x78) # zlib header + roundtripped = zlib.decompress(call["data"]).decode("utf-8") + self.assertNotIn("api_key", json.loads(roundtripped)) + + def test_zstd_sets_encoding_header_and_emits_standard_frame(self) -> None: + call = self._post(_results_response({}), compression=CaptureCompression.ZSTD) + self.assertEqual(call["headers"]["Content-Encoding"], "zstd") + self.assertIsInstance(call["data"], bytes) + self.assertEqual(call["data"][:4], b"\x28\xb5\x2f\xfd") # zstd frame magic + roundtripped = zstandard.ZstdDecompressor().decompress(call["data"]) + body = json.loads(roundtripped.decode("utf-8")) + self.assertNotIn("api_key", body) + self.assertEqual(len(body["batch"]), 1) + + def test_zstd_without_package_raises_actionable_error(self) -> None: + with mock.patch("posthog.capture_v1._zstandard", None): + with self.assertRaises(ValueError) as ctx: + self._post(_results_response({}), compression=CaptureCompression.ZSTD) + self.assertIn("posthog[zstd]", str(ctx.exception)) + + +class TestParseV1Response(unittest.TestCase): + def test_success_parses_results_with_details(self) -> None: + res = _FakeResponse( + 200, + json_body={"results": {"u-1": {"result": "drop", "details": "spam"}}}, + ) + parsed = _parse_v1_response(res) + self.assertTrue(parsed.is_success) + self.assertEqual(parsed.results["u-1"].result, "drop") + self.assertEqual(parsed.results["u-1"].details, "spam") + + @parameterized.expand( + [ + ("unparseable_body", _FakeResponse(200, raise_json=True)), + ("missing_results_key", _FakeResponse(200, json_body={"foo": 1})), + ] + ) + def test_success_with_bad_body_is_malformed(self, _name, res) -> None: + parsed = _parse_v1_response(res) + self.assertTrue(parsed.is_success) + self.assertTrue(parsed.malformed) + + @parameterized.expand( + [ + ("error_description", {"error_description": "bad batch"}, "bad batch"), + ("error", {"error": "validation_error"}, "validation_error"), + ("detail", {"detail": "nope"}, "nope"), + ] + ) + def test_error_message_extracted_from_body(self, _name, body, expected) -> None: + parsed = _parse_v1_response(_FakeResponse(400, json_body=body)) + self.assertFalse(parsed.is_success) + self.assertEqual(parsed.error_message, expected) + + def test_error_message_falls_back_to_text(self) -> None: + parsed = _parse_v1_response(_FakeResponse(400, raise_json=True, text="boom")) + self.assertEqual(parsed.error_message, "boom") + + @parameterized.expand([("numeric", "2", 2.0), ("absent", None, None)]) + def test_retry_after_header(self, _name, header_value, expected) -> None: + headers = {"Retry-After": header_value} if header_value is not None else {} + parsed = _parse_v1_response(_FakeResponse(503, headers=headers)) + self.assertEqual(parsed.retry_after, expected) + + +class TestSendV1Batch(unittest.TestCase): + """Drives ``_send_v1_batch`` with a stubbed ``_post_v1`` and no real sleeps.""" + + def setUp(self) -> None: + sleep_patch = mock.patch("posthog.capture_v1.time.sleep") + self.sleep = sleep_patch.start() + self.addCleanup(sleep_patch.stop) + + def _run(self, batch, outcomes, **kwargs): + stub = _PostV1Stub(outcomes) + with mock.patch("posthog.capture_v1._post_v1", stub): + _send_v1_batch("phc_key", "https://app.posthog.com", batch, **kwargs) + return stub + + def _run_expecting_error(self, batch, outcomes, **kwargs): + stub = _PostV1Stub(outcomes) + with mock.patch("posthog.capture_v1._post_v1", stub): + with self.assertRaises(CaptureV1Error) as ctx: + _send_v1_batch("phc_key", "https://app.posthog.com", batch, **kwargs) + return stub, ctx.exception + + def test_all_ok_sends_once(self) -> None: + stub = self._run([_msg("u-1")], [_results_response({"u-1": "ok"})]) + self.assertEqual(len(stub.calls), 1) + self.sleep.assert_not_called() + + def test_absent_uuid_treated_as_accepted(self) -> None: + # Empty results map: the event is neither retried nor errored. + stub = self._run([_msg("u-1")], [_results_response({})]) + self.assertEqual(len(stub.calls), 1) + + def test_partial_retry_resends_only_retry_events(self) -> None: + batch = [_msg("u-ok"), _msg("u-retry")] + stub = self._run( + batch, + [ + _results_response({"u-ok": "ok", "u-retry": "retry"}), + _results_response({"u-retry": "ok"}), + ], + ) + self.assertEqual(len(stub.calls), 2) + self.assertEqual(stub.calls[0]["uuids"], ["u-ok", "u-retry"]) + # Second attempt carries only the event the server asked to retry. + self.assertEqual(stub.calls[1]["uuids"], ["u-retry"]) + + def test_request_id_and_created_at_stable_attempt_increments(self) -> None: + stub = self._run( + [_msg("u-1")], + [ + _results_response({"u-1": "retry"}), + _results_response({"u-1": "ok"}), + ], + ) + self.assertEqual(stub.calls[0]["request_id"], stub.calls[1]["request_id"]) + # created_at is hoisted once, so the envelope timestamp is identical + # across retry attempts (only the attempt header increments). + self.assertEqual(stub.calls[0]["created_at"], stub.calls[1]["created_at"]) + self.assertEqual([c["attempt"] for c in stub.calls], [1, 2]) + + def test_compression_forwarded_to_post_v1(self) -> None: + stub = self._run( + [_msg("u-1")], + [_results_response({"u-1": "ok"})], + compression=CaptureCompression.DEFLATE, + ) + self.assertEqual(stub.calls[0]["compression"], CaptureCompression.DEFLATE) + + def test_drop_on_2xx_surfaces_via_error(self) -> None: + # A server-chosen drop is terminal: even on an all-ok-otherwise 2xx with + # no retry events, the send raises so on_error sees the dropped uuid + # (matches posthog-go/posthog-rs — a 2xx is not full delivery). + batch = [_msg("u-ok"), _msg("u-drop")] + stub, exc = self._run_expecting_error( + batch, + [_results_response({"u-ok": "ok", "u-drop": ("drop", "invalid")})], + ) + self.assertEqual(len(stub.calls), 1) # terminal, not retried + self.assertEqual(exc.status, 200) + self.assertEqual(exc.drops, [("u-drop", "invalid")]) + self.assertEqual(exc.retry_exhausted, []) + + def test_all_ok_no_drops_does_not_raise(self) -> None: + # The success path is unchanged when the server drops nothing. + stub = self._run( + [_msg("u-1"), _msg("u-2")], + [_results_response({"u-1": "ok", "u-2": "warning"})], + ) + self.assertEqual(len(stub.calls), 1) + + def test_drop_accumulated_across_attempts_surfaces_on_later_success(self) -> None: + # Attempt 1 drops one event and retries another; attempt 2 clears the + # retry. The earlier drop must still surface — it is not lost when the + # outstanding retries succeed on a later 2xx. + batch = [_msg("u-drop"), _msg("u-retry")] + stub, exc = self._run_expecting_error( + batch, + [ + _results_response({"u-drop": ("drop", "billing"), "u-retry": "retry"}), + _results_response({"u-retry": "ok"}), + ], + ) + self.assertEqual(len(stub.calls), 2) + self.assertEqual(stub.calls[1]["uuids"], ["u-retry"]) # only retry resent + self.assertEqual(exc.drops, [("u-drop", "billing")]) + self.assertEqual(exc.retry_exhausted, []) + + def test_retry_exhausted_raises_with_uuids(self) -> None: + stub, exc = self._run_expecting_error( + [_msg("u-1")], + [_results_response({"u-1": "retry"}), _results_response({"u-1": "retry"})], + max_retries=1, + ) + self.assertEqual(len(stub.calls), 2) + self.assertEqual(exc.retry_exhausted, ["u-1"]) + self.assertEqual(exc.drops, []) + + def test_retry_exhausted_carries_earlier_drops(self) -> None: + # A drop seen on attempt 1 rides along on the retry-exhaustion error. + batch = [_msg("u-drop"), _msg("u-retry")] + stub, exc = self._run_expecting_error( + batch, + [ + _results_response({"u-drop": ("drop", "billing"), "u-retry": "retry"}), + _results_response({"u-retry": "retry"}), + ], + max_retries=1, + ) + self.assertEqual(len(stub.calls), 2) + self.assertEqual(exc.retry_exhausted, ["u-retry"]) + self.assertEqual(exc.drops, [("u-drop", "billing")]) + + def test_malformed_2xx_is_terminal(self) -> None: + stub, exc = self._run_expecting_error( + [_msg("u-1")], [_FakeResponse(200, raise_json=True)] + ) + self.assertEqual(len(stub.calls), 1) + self.assertEqual(exc.status, 200) + + @parameterized.expand([("bad_request", 400), ("rate_limited", 429)]) + def test_terminal_status_raises_immediately(self, _name, status) -> None: + stub, exc = self._run_expecting_error( + [_msg("u-1")], + [_FakeResponse(status, json_body={"error": "nope"})], + max_retries=2, + ) + self.assertEqual(len(stub.calls), 1) # not retried + self.assertEqual(exc.status, status) + + def test_retryable_status_then_success(self) -> None: + stub = self._run( + [_msg("u-1")], + [ + _FakeResponse(503, headers={"Retry-After": "2"}), + _results_response({"u-1": "ok"}), + ], + ) + self.assertEqual(len(stub.calls), 2) + self.sleep.assert_called_once_with(2.0) # honored Retry-After + + def test_retryable_status_exhausted_raises(self) -> None: + stub, exc = self._run_expecting_error( + [_msg("u-1")], [_FakeResponse(503), _FakeResponse(503)], max_retries=1 + ) + self.assertEqual(len(stub.calls), 2) + self.assertEqual(exc.status, 503) + + def test_transport_error_then_success(self) -> None: + stub = self._run( + [_msg("u-1")], + [ConnectionError("boom"), _results_response({"u-1": "ok"})], + ) + self.assertEqual(len(stub.calls), 2) + + def test_transport_error_exhausted_reraises_original(self) -> None: + stub = _PostV1Stub([ConnectionError("boom"), ConnectionError("boom")]) + with mock.patch("posthog.capture_v1._post_v1", stub): + with self.assertRaises(ConnectionError): + _send_v1_batch( + "phc_key", "https://app.posthog.com", [_msg("u-1")], max_retries=1 + ) + self.assertEqual(len(stub.calls), 2) + + def test_small_retry_after_does_not_shorten_backoff(self) -> None: + # A Retry-After smaller than the configured backoff must not make the + # client retry earlier than its own schedule (Retry-After is a minimum). + # attempt_index=1 -> configured backoff 2s; Retry-After 0.5s is ignored. + stub = self._run( + [_msg("u-1")], + [ + _results_response({"u-1": "retry"}), + _results_response({"u-1": "retry"}, headers={"Retry-After": "0.5"}), + _results_response({"u-1": "ok"}), + ], + max_retries=3, + ) + self.assertEqual(len(stub.calls), 3) + # First backoff (attempt_index 0) waits 1s; second (attempt_index 1) + # keeps the 2s configured backoff rather than the smaller 0.5s header. + self.assertEqual([c.args[0] for c in self.sleep.call_args_list], [1, 2]) + + +class TestBackoff(unittest.TestCase): + """Directly exercises ``_backoff``'s Retry-After-as-minimum + cap policy.""" + + @parameterized.expand( + [ + # (attempt_index, retry_after, expected sleep seconds) + ("first_no_header", 0, None, 1), + ("second_no_header", 1, None, 2), + ("exp_capped_at_30", 10, None, 30), + ("zero_header_uses_backoff", 0, 0, 1), + ("larger_header_wins", 0, 5.0, 5.0), + ("smaller_header_ignored", 3, 2.0, 8), # configured 8 > 2.0 + ("equal_header_and_backoff", 0, 1.0, 1), + ("header_at_ceiling", 0, 30.0, 30), + ("header_above_ceiling_clamped", 0, 120.0, _MAX_BACKOFF_SECONDS), + ("absurd_header_clamped", 0, 10**9, _MAX_BACKOFF_SECONDS), + ] + ) + def test_backoff(self, _name, attempt_index, retry_after, expected) -> None: + with mock.patch("posthog.capture_v1.time.sleep") as sleep: + _backoff(attempt_index, retry_after) + sleep.assert_called_once_with(expected) diff --git a/posthog/test/test_client.py b/posthog/test/test_client.py index 98bb4914..97ec6dfa 100644 --- a/posthog/test/test_client.py +++ b/posthog/test/test_client.py @@ -8,6 +8,7 @@ from unittest import mock from parameterized import parameterized +from posthog.capture_compression import CaptureCompression from posthog.client import Client from posthog.contexts import get_context_session_id, new_context, set_context_session from posthog.request import APIError, GetResponse @@ -3260,3 +3261,98 @@ def test_debug_flag_re_raises_exceptions(self, mock_enqueue): with self.assertRaises(Exception) as cm: method(*args, **kwargs) self.assertEqual(str(cm.exception), "Expected error") + + +class TestClientSyncCaptureMode(unittest.TestCase): + """Sync-mode `_enqueue` selects the analytics submitter by `capture_mode`; + the dedicated AI endpoint always uses the legacy submitter.""" + + def _client(self, **kwargs): + return Client(FAKE_TEST_API_KEY, sync_mode=True, **kwargs) + + @parameterized.expand( + [ + ("v0", None, False), + ("v1", "v1", True), + ] + ) + def test_capture_mode_selects_sync_submitter(self, _name, capture_mode, expects_v1): + kwargs = {"capture_mode": capture_mode} if capture_mode else {} + with ( + mock.patch("posthog.client.batch_post") as mock_post, + mock.patch("posthog.client._send_v1_batch") as mock_v1, + ): + self._client(**kwargs).capture("evt", distinct_id="d") + if expects_v1: + mock_post.assert_not_called() + mock_v1.assert_called_once() + sent_batch = mock_v1.call_args.args[2] + self.assertEqual(len(sent_batch), 1) + self.assertEqual(sent_batch[0]["event"], "evt") + else: + mock_v1.assert_not_called() + mock_post.assert_called_once() + + def test_v1_sync_forwards_config_to_submitter(self): + with ( + mock.patch("posthog.client.batch_post"), + mock.patch("posthog.client._send_v1_batch") as mock_v1, + ): + self._client( + capture_mode="v1", + capture_compression=CaptureCompression.GZIP, + max_retries=4, + historical_migration=True, + ).capture("evt", distinct_id="d") + kwargs = mock_v1.call_args.kwargs + self.assertEqual(kwargs["compression"], CaptureCompression.GZIP) + self.assertEqual(kwargs["max_retries"], 4) + self.assertEqual(kwargs["historical_migration"], True) + + def test_v1_sync_gzip_flag_falls_back_to_gzip_compression(self): + # Legacy `gzip=True` with no explicit capture_compression -> GZIP on v1. + with ( + mock.patch("posthog.client.batch_post"), + mock.patch("posthog.client._send_v1_batch") as mock_v1, + ): + self._client(capture_mode="v1", gzip=True).capture("evt", distinct_id="d") + self.assertEqual( + mock_v1.call_args.kwargs["compression"], CaptureCompression.GZIP + ) + + def test_v1_sync_dedicated_ai_event_stays_legacy(self): + # $ai_* on the dedicated AI endpoint has no v1 form. + with ( + mock.patch("posthog.client.batch_post") as mock_post, + mock.patch("posthog.client._send_v1_batch") as mock_v1, + ): + client = self._client(capture_mode="v1", _dedicated_ai_endpoint=True) + client.capture("$ai_generation", distinct_id="d") + mock_v1.assert_not_called() + mock_post.assert_called_once() + self.assertEqual(mock_post.call_args.kwargs["path"], "/i/v0/ai/batch/") + + def test_v1_sync_dedicated_ai_analytics_event_uses_v1(self): + with ( + mock.patch("posthog.client.batch_post") as mock_post, + mock.patch("posthog.client._send_v1_batch") as mock_v1, + ): + client = self._client(capture_mode="v1", _dedicated_ai_endpoint=True) + client.capture("regular_event", distinct_id="d") + mock_post.assert_not_called() + mock_v1.assert_called_once() + + def test_v1_sync_ai_event_uses_v1_without_dedicated_endpoint(self): + # Without the dedicated AI endpoint, $ai_* events follow `capture_mode` + # and ride the v1 submitter like any analytics event. + with ( + mock.patch("posthog.client.batch_post") as mock_post, + mock.patch("posthog.client._send_v1_batch") as mock_v1, + ): + client = self._client(capture_mode="v1") + client.capture("$ai_generation", distinct_id="d") + mock_post.assert_not_called() + mock_v1.assert_called_once() + sent_batch = mock_v1.call_args.args[2] + self.assertEqual(len(sent_batch), 1) + self.assertEqual(sent_batch[0]["event"], "$ai_generation") diff --git a/posthog/test/test_consumer.py b/posthog/test/test_consumer.py index c136e2b3..35d3db8a 100644 --- a/posthog/test/test_consumer.py +++ b/posthog/test/test_consumer.py @@ -11,8 +11,10 @@ except ImportError: from Queue import Queue +from posthog.capture_compression import CaptureCompression +from posthog.capture_mode import CaptureMode from posthog.consumer import MAX_MSG_SIZE, Consumer -from posthog.request import APIError +from posthog.request import AI_EVENTS_ENDPOINT, EVENTS_ENDPOINT, APIError from posthog.test.logging_helpers import capture_message_only_logs from posthog.test.test_utils import TEST_API_KEY @@ -268,3 +270,114 @@ def on_error(e: Exception, batch: list[dict[str, str]]) -> None: self.assertEqual(len(on_error_called), 1) self.assertEqual(str(on_error_called[0][0]), "request failed") self.assertEqual(on_error_called[0][1], [track]) + + +def _ai_event(event_name: str = "$ai_generation") -> dict[str, str]: + return {"type": "track", "event": event_name, "distinct_id": "distinct_id"} + + +class TestConsumerCaptureModeRouting(unittest.TestCase): + """`capture_mode` selects the analytics submitter; the dedicated AI endpoint + has no v1 form and always uses the legacy submitter.""" + + @parameterized.expand( + [ + ("v0", CaptureMode.V0, False), + ("v1", CaptureMode.V1, True), + ] + ) + def test_capture_mode_selects_analytics_submitter( + self, _name, mode, expects_v1 + ) -> None: + consumer = Consumer(None, TEST_API_KEY, capture_mode=mode) + batch = [_track_event()] + with ( + mock.patch("posthog.consumer.batch_post") as mock_post, + mock.patch("posthog.consumer._send_v1_batch") as mock_v1, + ): + consumer.request(batch) + if expects_v1: + mock_post.assert_not_called() + mock_v1.assert_called_once() + self.assertEqual(mock_v1.call_args.args[2], batch) + else: + mock_v1.assert_not_called() + mock_post.assert_called_once() + self.assertEqual(mock_post.call_args.kwargs["path"], EVENTS_ENDPOINT) + + def test_v1_forwards_consumer_config_to_submitter(self) -> None: + consumer = Consumer( + None, + TEST_API_KEY, + capture_mode=CaptureMode.V1, + capture_compression=CaptureCompression.DEFLATE, + timeout=7, + retries=4, + historical_migration=True, + ) + with ( + mock.patch("posthog.consumer.batch_post"), + mock.patch("posthog.consumer._send_v1_batch") as mock_v1, + ): + consumer.request([_track_event()]) + kwargs = mock_v1.call_args.kwargs + self.assertEqual(kwargs["compression"], CaptureCompression.DEFLATE) + self.assertEqual(kwargs["timeout"], 7) + self.assertEqual(kwargs["max_retries"], 4) + self.assertEqual(kwargs["historical_migration"], True) + + def test_v1_dedicated_ai_splits_submitters(self) -> None: + # Analytics -> v1 submitter; $ai_* -> legacy AI endpoint. + consumer = Consumer( + None, + TEST_API_KEY, + capture_mode=CaptureMode.V1, + dedicated_ai_endpoint=True, + ) + analytics, ai = _track_event(), _ai_event() + with ( + mock.patch("posthog.consumer.batch_post") as mock_post, + mock.patch("posthog.consumer._send_v1_batch") as mock_v1, + ): + consumer.request([analytics, ai]) + mock_v1.assert_called_once() + self.assertEqual(mock_v1.call_args.args[2], [analytics]) + mock_post.assert_called_once() + self.assertEqual(mock_post.call_args.kwargs["path"], AI_EVENTS_ENDPOINT) + self.assertEqual(mock_post.call_args.kwargs["batch"], [ai]) + + def test_v1_dedicated_ai_only_ai_events_skips_v1_submitter(self) -> None: + consumer = Consumer( + None, + TEST_API_KEY, + capture_mode=CaptureMode.V1, + dedicated_ai_endpoint=True, + ) + with ( + mock.patch("posthog.consumer.batch_post") as mock_post, + mock.patch("posthog.consumer._send_v1_batch") as mock_v1, + ): + consumer.request([_ai_event()]) + mock_v1.assert_not_called() + mock_post.assert_called_once() + self.assertEqual(mock_post.call_args.kwargs["path"], AI_EVENTS_ENDPOINT) + + def test_v1_without_dedicated_ai_endpoint_routes_ai_events_through_v1(self) -> None: + # The dedicated AI endpoint is the only v1 gate: with it off, $ai_* events + # follow `capture_mode` like any analytics event and ride the v1 submitter + # (matching legacy, where they ship to the standard analytics endpoint). + consumer = Consumer( + None, + TEST_API_KEY, + capture_mode=CaptureMode.V1, + dedicated_ai_endpoint=False, + ) + batch = [_ai_event(), _track_event()] + with ( + mock.patch("posthog.consumer.batch_post") as mock_post, + mock.patch("posthog.consumer._send_v1_batch") as mock_v1, + ): + consumer.request(batch) + mock_v1.assert_called_once() + self.assertEqual(mock_v1.call_args.args[2], batch) + mock_post.assert_not_called() diff --git a/pyproject.toml b/pyproject.toml index d809ddfe..565baf75 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,11 @@ Repository = "https://github.com/posthog/posthog-python" [project.optional-dependencies] langchain = ["langchain>=0.2.0"] +# Opt-in zstd support for capture-v1 request compression. Kept out of the core +# dependencies so existing installs are unaffected; Python gains stdlib zstd +# only in 3.14 (compression.zstd), so the third-party package is needed until +# then. +zstd = ["zstandard>=0.23.0"] # Note: the MCP SDK (`mcp`) is intentionally NOT an extra. It's a peer dependency of # `instrument()` — anyone wrapping a FastMCP/Server already has it — so it's imported # lazily and version-checked at runtime, not installed by posthog. `PostHogMCP` (custom @@ -89,6 +94,7 @@ test = [ "opentelemetry-sdk>=1.20.0", "opentelemetry-exporter-otlp-proto-http>=1.20.0", "pytest-bdd>=8.1.0", + "zstandard>=0.23.0", ] [tool.setuptools] diff --git a/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index ef176a91..d71198ba 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -4,6 +4,8 @@ # members. Modules with __all__ use it; other modules include non-underscore # names. External imports are excluded. alias posthog.BeforeSendCallback -> posthog.types.BeforeSendCallback +alias posthog.CaptureCompression -> posthog.capture_compression.CaptureCompression +alias posthog.CaptureMode -> posthog.capture_mode.CaptureMode alias posthog.Client -> posthog.client.Client alias posthog.DEFAULT_CODE_VARIABLES_DETECT_SECRETS -> posthog.exception_utils.DEFAULT_CODE_VARIABLES_DETECT_SECRETS alias posthog.DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS -> posthog.exception_utils.DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS @@ -216,6 +218,8 @@ alias posthog.args.FeatureFlagEvaluations -> posthog.feature_flag_evaluations.Fe alias posthog.args.SendFeatureFlagsOptions -> posthog.types.SendFeatureFlagsOptions alias posthog.client.AI_EVENTS_ENDPOINT -> posthog.request.AI_EVENTS_ENDPOINT alias posthog.client.APIError -> posthog.request.APIError +alias posthog.client.CaptureCompression -> posthog.capture_compression.CaptureCompression +alias posthog.client.CaptureMode -> posthog.capture_mode.CaptureMode alias posthog.client.Consumer -> posthog.consumer.Consumer alias posthog.client.DEFAULT_CODE_VARIABLES_DETECT_SECRETS -> posthog.exception_utils.DEFAULT_CODE_VARIABLES_DETECT_SECRETS alias posthog.client.DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS -> posthog.exception_utils.DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS @@ -282,6 +286,8 @@ alias posthog.client.to_values -> posthog.types.to_values alias posthog.client.try_attach_code_variables_to_frames -> posthog.exception_utils.try_attach_code_variables_to_frames alias posthog.consumer.AI_EVENTS_ENDPOINT -> posthog.request.AI_EVENTS_ENDPOINT alias posthog.consumer.APIError -> posthog.request.APIError +alias posthog.consumer.CaptureCompression -> posthog.capture_compression.CaptureCompression +alias posthog.consumer.CaptureMode -> posthog.capture_mode.CaptureMode alias posthog.consumer.DatetimeSerializer -> posthog.request.DatetimeSerializer alias posthog.consumer.EVENTS_ENDPOINT -> posthog.request.EVENTS_ENDPOINT alias posthog.consumer.batch_post -> posthog.request.batch_post @@ -470,9 +476,23 @@ attribute posthog.before_send = None attribute posthog.bucketed_rate_limiter.Number = Union[int, float] attribute posthog.bucketed_rate_limiter.ONE_DAY_IN_SECONDS = 86400.0 attribute posthog.bucketed_rate_limiter.log = logging.getLogger('posthog') +attribute posthog.capture_compression.CAPTURE_COMPRESSION_ENV_VAR = 'POSTHOG_CAPTURE_COMPRESSION' +attribute posthog.capture_compression.CaptureCompression.DEFLATE = 'deflate' +attribute posthog.capture_compression.CaptureCompression.GZIP = 'gzip' +attribute posthog.capture_compression.CaptureCompression.NONE = 'none' +attribute posthog.capture_compression.CaptureCompression.ZSTD = 'zstd' attribute posthog.capture_exception_code_variables = False +attribute posthog.capture_mode.CAPTURE_MODE_ENV_VAR = 'POSTHOG_CAPTURE_MODE' +attribute posthog.capture_mode.CaptureMode.V0 = 'v0' +attribute posthog.capture_mode.CaptureMode.V1 = 'v1' +attribute posthog.capture_v1.CaptureV1Error.attempts = attempts +attribute posthog.capture_v1.CaptureV1Error.drops = drops or [] +attribute posthog.capture_v1.CaptureV1Error.request_id = request_id +attribute posthog.capture_v1.CaptureV1Error.retry_exhausted = retry_exhausted or [] attribute posthog.client.Client.api_key = (project_api_key or '').strip() +attribute posthog.client.Client.capture_compression = _resolve_capture_compression(capture_compression, gzip_fallback=gzip) attribute posthog.client.Client.capture_exception_code_variables = capture_exception_code_variables +attribute posthog.client.Client.capture_mode = _resolve_capture_mode(capture_mode) attribute posthog.client.Client.code_variables_detect_secrets = code_variables_detect_secrets if code_variables_detect_secrets is not None else DEFAULT_CODE_VARIABLES_DETECT_SECRETS attribute posthog.client.Client.code_variables_ignore_patterns = code_variables_ignore_patterns if code_variables_ignore_patterns is not None else DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS attribute posthog.client.Client.code_variables_mask_patterns = code_variables_mask_patterns if code_variables_mask_patterns is not None else DEFAULT_CODE_VARIABLES_MASK_PATTERNS @@ -505,6 +525,7 @@ attribute posthog.client.Client.in_app_modules = in_app_modules attribute posthog.client.Client.is_server = is_server attribute posthog.client.Client.log = logging.getLogger('posthog') attribute posthog.client.Client.log_captured_exceptions = log_captured_exceptions +attribute posthog.client.Client.max_retries = max_retries attribute posthog.client.Client.on_error = on_error attribute posthog.client.Client.personal_api_key = (personal_api_key.strip() if isinstance(personal_api_key, str) else personal_api_key) or None attribute posthog.client.Client.poll_interval = poll_interval @@ -525,6 +546,8 @@ attribute posthog.code_variables_mask_url_credentials = DEFAULT_CODE_VARIABLES_M attribute posthog.consumer.AI_MAX_MSG_SIZE = 8 * 1024 * 1024 attribute posthog.consumer.BATCH_SIZE_LIMIT = 5 * 1024 * 1024 attribute posthog.consumer.Consumer.api_key = api_key +attribute posthog.consumer.Consumer.capture_compression = capture_compression +attribute posthog.consumer.Consumer.capture_mode = capture_mode attribute posthog.consumer.Consumer.daemon = True attribute posthog.consumer.Consumer.dedicated_ai_endpoint = dedicated_ai_endpoint attribute posthog.consumer.Consumer.flush_at = flush_at @@ -827,8 +850,11 @@ class posthog.ai.types.ToolInProgress class posthog.args.OptionalCaptureArgs class posthog.args.OptionalSetArgs class posthog.bucketed_rate_limiter.BucketedRateLimiter(bucket_size: Number, refill_rate: Number, refill_interval_seconds: Number, on_bucket_rate_limited: Optional[Callable[[Hashable], None]] = None, clock: Callable[[], float] = time.monotonic) -class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, _dedicated_ai_endpoint=False) -class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, gzip=False, retries=10, timeout=15, historical_migration=False, dedicated_ai_endpoint=False) +class posthog.capture_compression.CaptureCompression +class posthog.capture_mode.CaptureMode +class posthog.capture_v1.CaptureV1Error(status: int | str, message: str, *, retry_after: Optional[float] = None, request_id: Optional[str] = None, attempts: Optional[int] = None, retry_exhausted: Optional[list[str]] = None, drops: Optional[list[tuple[str, Optional[str]]]] = None) +class posthog.client.Client(project_api_key: str, host=None, debug=False, max_queue_size=10000, send=True, on_error=None, flush_at=100, flush_interval=5.0, gzip=False, max_retries=3, sync_mode=False, timeout=15, thread=1, poll_interval=30, personal_api_key=None, disabled=False, disable_geoip=True, is_server=True, historical_migration=False, feature_flags_request_timeout_seconds=3, feature_flags_request_max_retries=1, super_properties=None, enable_exception_autocapture=False, log_captured_exceptions=False, project_root=None, privacy_mode=False, before_send=None, flag_fallback_cache_url=None, enable_local_evaluation=True, flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None, capture_exception_code_variables=False, code_variables_mask_patterns=None, code_variables_ignore_patterns=None, code_variables_mask_url_credentials=None, code_variables_detect_secrets=None, in_app_modules: list[str] | None = None, enable_exception_autocapture_rate_limiting=False, exception_autocapture_bucket_size=ExceptionCapture.DEFAULT_BUCKET_SIZE, exception_autocapture_refill_rate=ExceptionCapture.DEFAULT_REFILL_RATE, exception_autocapture_refill_interval_seconds=ExceptionCapture.DEFAULT_REFILL_INTERVAL_SECONDS, capture_mode: Optional[Union[CaptureMode, str]] = None, capture_compression: Optional[Union[CaptureCompression, str]] = None, _dedicated_ai_endpoint=False) +class posthog.consumer.Consumer(queue, api_key, flush_at=100, host=None, on_error=None, flush_interval=5.0, gzip=False, retries=10, timeout=15, historical_migration=False, dedicated_ai_endpoint=False, capture_mode=CaptureMode.V0, capture_compression=CaptureCompression.NONE) class posthog.contexts.ContextScope(parent=None, fresh: bool = False, capture_exceptions: bool = True, client: Optional[Client] = None) class posthog.exception_capture.ExceptionCapture(client: Client, rate_limiting_enabled=False, bucket_size=DEFAULT_BUCKET_SIZE, refill_rate=DEFAULT_REFILL_RATE, refill_interval_seconds=DEFAULT_REFILL_INTERVAL_SECONDS) class posthog.exception_utils.AnnotatedValue(value, metadata) @@ -1297,6 +1323,9 @@ module posthog.ai.types module posthog.ai.utils module posthog.args module posthog.bucketed_rate_limiter +module posthog.capture_compression +module posthog.capture_mode +module posthog.capture_v1 module posthog.client module posthog.consumer module posthog.contexts diff --git a/sdk_compliance_adapter/Dockerfile.v1 b/sdk_compliance_adapter/Dockerfile.v1 new file mode 100644 index 00000000..6891837a --- /dev/null +++ b/sdk_compliance_adapter/Dockerfile.v1 @@ -0,0 +1,25 @@ +FROM python:3.12-slim + +WORKDIR /app + +# Copy the SDK source code +COPY posthog/ /app/sdk/posthog/ +COPY setup.py pyproject.toml README.md LICENSE /app/sdk/ + +# Install the SDK from source +RUN cd /app/sdk && pip install --no-cache-dir -e . + +# Install adapter dependencies +RUN pip install --no-cache-dir flask python-dateutil + +# Copy adapter code +COPY sdk_compliance_adapter/adapter.py /app/adapter.py + +# Select the capture-v1 protocol; same adapter code, different runtime mode. +ENV CAPTURE_MODE=v1 + +# Expose port 8080 +EXPOSE 8080 + +# Run the adapter +CMD ["python", "/app/adapter.py"] diff --git a/sdk_compliance_adapter/adapter.py b/sdk_compliance_adapter/adapter.py index 88dbb691..d9453e00 100644 --- a/sdk_compliance_adapter/adapter.py +++ b/sdk_compliance_adapter/adapter.py @@ -14,6 +14,8 @@ from flask import Flask, jsonify, request from posthog import Client +from posthog.capture_compression import CaptureCompression +from posthog.capture_v1 import _post_v1 as original_post_v1 from posthog.request import EVENTS_ENDPOINT from posthog.request import batch_post as original_batch_post from posthog.version import VERSION @@ -26,6 +28,16 @@ app = Flask(__name__) +# Selects which capture protocol this adapter process speaks. Baked at build +# time via the CAPTURE_MODE env var ("v1" => capture-v1, anything else => legacy +# v0), mirroring the v0/v1 Dockerfile split. One process speaks one mode and +# advertises it via /health capabilities. +CAPTURE_MODE = os.environ.get("CAPTURE_MODE", "") + + +def is_v1() -> bool: + return CAPTURE_MODE == "v1" + class RequestInfo: """Information about an HTTP request made by the SDK""" @@ -132,6 +144,33 @@ def record_request(self, status_code: int, batch: List[Dict], batch_id: str): if retry_attempt > 0: self.total_retries += 1 + def record_request_v1( + self, status_code: int, batch: List[Dict], attempt: int, terminal_count: int + ): + """Record a capture-v1 HTTP attempt. + + Unlike v0, the retry attempt is carried on the request (PostHog-Attempt, + 1-based) rather than inferred from a batch id, and a 2xx no longer means + the whole batch was accepted — only events with a terminal (non-"retry") + per-event result count as sent. + """ + with self.lock: + uuid_list = [event.get("uuid", "") for event in batch] + self.requests_made.append( + RequestInfo( + timestamp_ms=int(time.time() * 1000), + status_code=status_code, + retry_attempt=attempt - 1, + event_count=len(batch), + uuid_list=uuid_list, + ) + ) + if attempt > 1: + self.total_retries += 1 + if 200 <= status_code < 300: + self.total_events_sent += terminal_count + self.pending_events = max(0, self.pending_events - terminal_count) + def record_error(self, error: str): """Record an error""" with self.lock: @@ -188,6 +227,60 @@ def patched_batch_post( raise +def patched_post_v1( + api_key: str, + host: Optional[str], + batch_body: Dict, + *, + attempt: int, + request_id: str, + compression: CaptureCompression = CaptureCompression.NONE, + timeout: int = 15, + session: Any = None, +): + """Patched version of _post_v1 that records requests for /state assertions. + + Mirrors the legacy `patched_batch_post`, but reads the retry attempt from the + call (1-based) and counts only terminal per-event results as sent. + """ + batch = batch_body.get("batch", []) + try: + response = original_post_v1( + api_key, + host, + batch_body, + attempt=attempt, + request_id=request_id, + compression=compression, + timeout=timeout, + session=session, + ) + except Exception as e: + status_code = getattr(e, "status", 0) + state.record_request_v1( + status_code if isinstance(status_code, int) else 0, batch, attempt, 0 + ) + state.record_error(str(e)) + raise + + terminal = 0 + status = response.status_code + if 200 <= status < 300: + try: + results = response.json().get("results", {}) + # Mirror _send_v1_batch: only a non-retry directive is terminal. A + # missing/null `result` is not counted as sent. + terminal = sum( + 1 + for r in results.values() + if (r or {}).get("result") not in (None, "retry") + ) + except Exception: + terminal = 0 + state.record_request_v1(status, batch, attempt, terminal) + return response + + # Monkey-patch the batch_post function import posthog.request # noqa: E402 @@ -198,16 +291,26 @@ def patched_batch_post( posthog.consumer.batch_post = patched_batch_post +# Patch the capture-v1 submitter. `_send_v1_batch` resolves `_post_v1` as a module +# global at call time, so patching it here covers both the async consumer and the +# sync client paths. +import posthog.capture_v1 # noqa: E402 + +posthog.capture_v1._post_v1 = patched_post_v1 + @app.route("/health", methods=["GET"]) def health(): """Health check endpoint""" + capabilities = ( + ["capture_v1", "encoding_gzip"] if is_v1() else ["capture_v0", "encoding_gzip"] + ) return jsonify( { "sdk_name": "posthog-python", "sdk_version": VERSION, "adapter_version": "1.0.0", - "capabilities": ["capture_v0", "encoding_gzip"], + "capabilities": capabilities, } ) @@ -228,6 +331,11 @@ def init(): flush_interval_ms = data.get("flush_interval_ms", 500) max_retries = data.get("max_retries", 3) enable_compression = data.get("enable_compression", False) + # Compliance tests assert the request-level default when callers omit + # disable_geoip, so the adapter default keeps geoip-enabled /flags + # requests while still allowing per-call overrides. + disable_geoip = data.get("disable_geoip", False) + historical_migration = data.get("historical_migration", False) if not api_key: return jsonify({"error": "api_key is required"}), 400 @@ -237,6 +345,9 @@ def init(): # Convert flush_interval from ms to seconds flush_interval = flush_interval_ms / 1000.0 + # One adapter process speaks one capture protocol, selected by CAPTURE_MODE. + capture_mode = "v1" if is_v1() else "v0" + # Create client client = Client( project_api_key=api_key, @@ -246,10 +357,9 @@ def init(): gzip=enable_compression, max_retries=max_retries, debug=False, - # Compliance tests assert the request-level default when callers omit - # disable_geoip. Configure the adapter to exercise geoip-enabled - # /flags requests by default while still allowing per-call overrides. - disable_geoip=False, + disable_geoip=disable_geoip, + historical_migration=historical_migration, + capture_mode=capture_mode, ) state.client = client @@ -257,7 +367,9 @@ def init(): logger.info( f"Initialized SDK with api_key={api_key[:10]}..., host={host}, " f"flush_at={flush_at}, flush_interval={flush_interval}, " - f"max_retries={max_retries}, gzip={enable_compression}" + f"max_retries={max_retries}, gzip={enable_compression}, " + f"capture_mode={capture_mode}, disable_geoip={disable_geoip}, " + f"historical_migration={historical_migration}" ) return jsonify({"success": True}) @@ -279,12 +391,28 @@ def capture(): event = data.get("event") properties = data.get("properties") timestamp = data.get("timestamp") + options = data.get("options") if not distinct_id: return jsonify({"error": "distinct_id is required"}), 400 if not event: return jsonify({"error": "event is required"}), 400 + # Fold capture-v1 options back into the magic `$`-prefixed properties the + # SDK lifts onto the wire `options` object. Renamed keys mirror the SDK's + # sentinel table; unknown keys get a bare `$` prefix. v0 has no wire + # options object, so this only applies in v1 mode. + if options and is_v1(): + properties = dict(properties or {}) + option_to_property = { + "cookieless_mode": "$cookieless_mode", + "disable_skew_correction": "$ignore_sent_at", + "process_person_profile": "$process_person_profile", + "product_tour_id": "$product_tour_id", + } + for key, value in options.items(): + properties[option_to_property.get(key, "$" + key)] = value + # Capture event kwargs = {"distinct_id": distinct_id, "properties": properties} if timestamp: diff --git a/sdk_compliance_adapter/docker-compose.yml b/sdk_compliance_adapter/docker-compose.yml index 471db84d..e6519de5 100644 --- a/sdk_compliance_adapter/docker-compose.yml +++ b/sdk_compliance_adapter/docker-compose.yml @@ -1,7 +1,7 @@ version: "3.8" services: - # PostHog Python SDK adapter + # PostHog Python SDK adapter (capture v0) sdk-adapter: build: context: .. @@ -11,6 +11,16 @@ services: networks: - test-network + # PostHog Python SDK adapter (capture v1) + sdk-adapter-v1: + build: + context: .. + dockerfile: sdk_compliance_adapter/Dockerfile.v1 + ports: + - "8082:8080" + networks: + - test-network + # Test harness test-harness: image: ghcr.io/posthog/sdk-test-harness:0.10.0 diff --git a/uv.lock b/uv.lock index 0fad4d84..12a0e737 100644 --- a/uv.lock +++ b/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. +exclude-newer = "2026-06-29T19:09:37.282668Z" exclude-newer-span = "P7D" [manifest] @@ -2538,6 +2538,10 @@ test = [ { name = "python-dateutil" }, { name = "tiktoken" }, { name = "tzdata" }, + { name = "zstandard" }, +] +zstd = [ + { name = "zstandard" }, ] [package.dev-dependencies] @@ -2596,8 +2600,10 @@ requires-dist = [ { name = "typing-extensions", specifier = ">=4.2.0" }, { name = "tzdata", marker = "extra == 'test'" }, { name = "wheel", marker = "extra == 'dev'" }, + { name = "zstandard", marker = "extra == 'test'", specifier = ">=0.23.0" }, + { name = "zstandard", marker = "extra == 'zstd'", specifier = ">=0.23.0" }, ] -provides-extras = ["langchain", "otel", "dev", "test"] +provides-extras = ["langchain", "zstd", "otel", "dev", "test"] [package.metadata.requires-dev] dev = [{ name = "claude-agent-sdk", specifier = ">=0.1.50" }]