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/client.py b/posthog/client.py index 6a9846cf..b5136b72 100644 --- a/posthog/client.py +++ b/posthog/client.py @@ -16,7 +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, @@ -262,6 +267,7 @@ def __init__( 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, ): """ @@ -350,6 +356,11 @@ def __init__( (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 @@ -377,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 ) @@ -409,6 +421,11 @@ def __init__( # `/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 @@ -515,6 +532,7 @@ def __init__( 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) @@ -1538,6 +1556,7 @@ def _reinit_after_fork(self): 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) @@ -1637,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 cf5256db..0abd24ca 100644 --- a/posthog/consumer.py +++ b/posthog/consumer.py @@ -5,7 +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, @@ -52,6 +54,7 @@ def __init__( historical_migration=False, dedicated_ai_endpoint=False, capture_mode=CaptureMode.V0, + capture_compression=CaptureCompression.NONE, ): """Create a consumer thread.""" Thread.__init__(self) @@ -66,6 +69,7 @@ def __init__( 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 @@ -154,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] = [] @@ -166,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 index 5dad1881..a4ce0154 100644 --- a/posthog/test/test_capture_compression.py +++ b/posthog/test/test_capture_compression.py @@ -9,7 +9,10 @@ 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): @@ -118,3 +121,48 @@ def test_env_zstd_without_package_warns_and_uses_fallback(self) -> None: 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_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/references/public_api_snapshot.txt b/references/public_api_snapshot.txt index 0319c187..d71198ba 100644 --- a/references/public_api_snapshot.txt +++ b/references/public_api_snapshot.txt @@ -218,6 +218,7 @@ 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 @@ -285,6 +286,7 @@ 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 @@ -488,6 +490,7 @@ 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 @@ -522,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 @@ -542,6 +546,7 @@ 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 @@ -848,8 +853,8 @@ class posthog.bucketed_rate_limiter.BucketedRateLimiter(bucket_size: Number, ref 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, _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) +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) 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