feat(analytics): add an early capture filter

.sampo/changesets/context-capture-filter.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add an optional `should_capture` callback to `Client`/`Posthog` and module settings for context-based analytics and AI event filtering. It receives no payload and must return `True` to allow capture. Rejected events skip payload cleaning; allowed events avoid the extra cleaning required by `before_send`. Existing `before_send` behavior is unchanged.

posthog/__init__.py

@@ -362,6 +362,11 @@ def get_tags() -> Dict[str, Any]:
     privacy_mode: Capture AI usage metadata without prompt inputs or outputs.
     before_send: Optional callback that can modify or drop events before upload.
         Return ``None`` to drop an event.
+    should_capture: Optional synchronous callback for context-based filtering of
+        analytics and AI events before payload cleaning. Receives no event and
+        must return ``True`` to allow capture; any other result or an exception
+        drops the event. Does not filter metrics or tracing spans. Use
+        ``before_send`` when filtering depends on event contents.
     enable_local_evaluation: Whether to poll feature flag definitions for local
         evaluation when a personal API key is configured.
     flag_definition_cache_provider: Optional external cache provider for sharing
@@ -420,6 +425,7 @@ def get_tags() -> Dict[str, Any]:
 # Used for our AI observability feature to not capture any prompt or output just usage + metadata
 privacy_mode = False  # type: bool
 before_send = None  # type: Optional[BeforeSendCallback]
+should_capture: Optional[Callable[[], bool]] = None
 # Whether to enable feature flag polling for local evaluation by default. Defaults to True.
 # 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
@@ -1410,6 +1416,7 @@ def setup() -> Client:
             project_root=project_root,
             privacy_mode=privacy_mode,
             before_send=before_send,
+            should_capture=should_capture,
             enable_local_evaluation=enable_local_evaluation,
             flag_definition_cache_provider=flag_definition_cache_provider,
             capture_exception_code_variables=capture_exception_code_variables,
@@ -1432,6 +1439,7 @@ def setup() -> Client:
     default_client.debug = debug
     default_client.privacy_mode = bool(privacy_mode)
     default_client._set_before_send(before_send)
+    default_client.should_capture = should_capture
     default_client.enable_full_ai_capture = (
         bool(enable_full_ai_capture)
         or bool(_use_ai_lane)

posthog/client.py

@@ -720,6 +720,7 @@ def __init__(
         _use_ai_lane=False,
         _enable_multimodal_capture=False,
         traces: Optional[dict] = None,
+        should_capture: Optional[Callable[[], bool]] = None,
     ):
         """
         Initialize a new PostHog client instance.
@@ -787,6 +788,12 @@ def __init__(
                 False.
             before_send: Optional callback that can modify or drop events before
                 upload. Return ``None`` to drop an event.
+            should_capture: Optional synchronous callback for context-based event
+                filtering. Runs in the caller's context before payload cleaning;
+                receives no event and must return ``True`` to allow capture.
+                Any other result or an exception drops the event. Applies to
+                analytics and AI events, not metrics or tracing spans. Use
+                ``before_send`` when filtering depends on event contents.
             flag_fallback_cache_url: Optional feature flag fallback cache URL,
                 such as ``memory://local/?ttl=300&size=10000`` or a Redis URL.
             enable_local_evaluation: Whether to poll feature flag definitions for
@@ -1049,6 +1056,7 @@ def __init__(
             )
 
         self._set_before_send(before_send)
+        self.should_capture = should_capture
 
         if self.enable_exception_autocapture:
             self.exception_capture = ExceptionCapture(
@@ -1674,6 +1682,10 @@ def _capture(
         # applied to the fully-enriched properties dict just before enqueueing.
         property_allowlist = kwargs.get("_property_allowlist", None)
 
+        # Checked before flag evaluation, so a rejected event makes no `/flags` request.
+        if not self._should_capture_event():
+            return None
+
         properties = {**(properties or {}), **system_context()}
 
         if self.capture_trace_context:
@@ -1790,7 +1802,11 @@ def _capture(
             msg["properties"] = properties
 
         return self._enqueue(
-            msg, disable_geoip, lane, property_allowlist=property_allowlist
+            msg,
+            disable_geoip,
+            lane,
+            property_allowlist=property_allowlist,
+            admitted=True,
         )
 
     def _parse_send_feature_flags(self, send_feature_flags) -> SendFeatureFlagsOptions:
@@ -2341,16 +2357,36 @@ def _normalize_event_uuid(self, msg):
             # Always send a uuid, so we can always return one
             msg["uuid"] = stringify_id(uuid4())
 
-    def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
+    def _should_capture_event(self) -> bool:
+        """Run the `should_capture` callback; any result but `True`, or an error, drops the event."""
+        if self.should_capture is None:
+            return True
+        try:
+            if self.should_capture() is True:
+                return True
+            self.log.debug("Event dropped by should_capture callback")
+        except Exception:
+            self.log.warning("Error in should_capture callback; dropping event")
+        return False
+
+    def _enqueue(
+        self, msg, disable_geoip, lane=None, property_allowlist=None, admitted=False
+    ):
         # type: (...) -> Optional[str]
-        """Push a new `msg` onto a lane's queue (analytics when unspecified), return the event uuid or None."""
+        """Push a new `msg` onto a lane's queue (analytics when unspecified), return the event uuid or None.
+
+        `admitted` means the caller already ran the `should_capture` callback.
+        """
 
         if lane is None:
             lane = self._analytics_lane
 
         if self.disabled:
             return None
 
+        if not admitted and not self._should_capture_event():
+            return None
+
         timestamp = msg["timestamp"]
         if timestamp is None:
             timestamp = datetime.now(tz=timezone.utc)
@@ -4214,15 +4250,17 @@ def _capture_feature_flag_called_if_needed(
                 _MINIMAL_FLAG_CALLED_EVENT_PROPERTIES
             )
 
-        self.capture(
+        event_uuid = self.capture(
             "$feature_flag_called",
             distinct_id=distinct_id,
             properties=properties,
             groups=groups or {},
             disable_geoip=disable_geoip,
             **extra_capture_kwargs,
         )
-        reported_flags.add(feature_flag_reported_key)
+        # A dropped event was not reported, so a later allowed call must still send it.
+        if event_uuid is not None:
+            reported_flags.add(feature_flag_reported_key)
 
     def get_remote_config_payload(self, key: str):
         """

posthog/test/test_should_capture.py

@@ -0,0 +1,213 @@
+import importlib
+import warnings
+from contextvars import ContextVar
+from concurrent.futures import ThreadPoolExecutor
+from unittest import mock
+
+import pytest
+
+import posthog
+from posthog.client import Client
+from posthog.test.test_utils import FAKE_TEST_API_KEY
+from posthog.utils import clean
+
+
+@pytest.mark.parametrize("method", ["capture", "capture_ai"])
+@pytest.mark.parametrize("should_capture", [None, lambda: True])
+def test_allowed_events_are_cleaned_once(method, should_capture):
+    client = Client(FAKE_TEST_API_KEY, should_capture=should_capture, sync_mode=True)
+    with (
+        mock.patch("posthog.client.clean", wraps=clean) as clean_payload,
+        mock.patch("posthog.client.batch_post") as post,
+    ):
+        event_uuid = getattr(client, method)(
+            "example_event", distinct_id="user1", properties={"nested": [{"value": 1}]}
+        )
+
+    assert event_uuid is not None
+    clean_payload.assert_called_once()
+    post.assert_called_once()
+    assert post.call_args.kwargs["batch"][0]["properties"]["nested"] == [{"value": 1}]
+
+
+@pytest.mark.parametrize("method", ["capture", "capture_ai"])
+@pytest.mark.parametrize("result", [False, None, 1, "true"])
+def test_rejected_events_skip_normalization_cleaning_and_before_send(method, result):
+    before_send = mock.Mock()
+    client = Client(
+        FAKE_TEST_API_KEY,
+        should_capture=lambda: result,
+        before_send=before_send,
+        sync_mode=True,
+    )
+    with (
+        mock.patch("posthog.client.clean") as clean_payload,
+        mock.patch.object(client, "_normalize_event_uuid") as normalize_uuid,
+        mock.patch("posthog.client.batch_post") as post,
+    ):
+        assert getattr(client, method)("example_event", distinct_id="user1") is None
+
+    clean_payload.assert_not_called()
+    normalize_uuid.assert_not_called()
+    before_send.assert_not_called()
+    post.assert_not_called()
+
+
+@pytest.mark.parametrize("allowed", [False, True])
+def test_feature_flags_are_requested_only_for_allowed_events(allowed):
+    should_capture = mock.Mock(return_value=allowed)
+    client = Client(FAKE_TEST_API_KEY, should_capture=should_capture, send=False)
+    with (
+        mock.patch("posthog.client.flags", return_value={"featureFlags": {}}) as flags,
+        warnings.catch_warnings(),
+    ):
+        warnings.simplefilter("ignore", DeprecationWarning)
+        event_uuid = client.capture(
+            "example_event", distinct_id="user1", send_feature_flags=True
+        )
+
+    assert (event_uuid is not None) is allowed
+    assert flags.call_count == int(allowed)
+    should_capture.assert_called_once_with()
+
+
+def test_callback_failure_drops_event_without_logging_its_contents(caplog):
+    should_capture = mock.Mock(side_effect=ValueError("private exception content"))
+    client = Client(FAKE_TEST_API_KEY, should_capture=should_capture, sync_mode=True)
+    with mock.patch("posthog.client.batch_post") as post:
+        assert client.capture("private event", distinct_id="private user") is None
+
+    post.assert_not_called()
+    assert "Error in should_capture callback; dropping event" in caplog.text
+    assert "private" not in caplog.text
+
+
+def test_allowed_events_still_run_and_reclean_mutating_before_send():
+    unsupported_value = object()
+
+    def before_send(event):
+        event["properties"]["unsupported"] = unsupported_value
+        event["properties"]["scrubbed"] = True
+        return event
+
+    client = Client(
+        FAKE_TEST_API_KEY,
+        should_capture=lambda: True,
+        before_send=before_send,
+        sync_mode=True,
+    )
+    with (
+        mock.patch("posthog.client.clean", wraps=clean) as clean_payload,
+        mock.patch("posthog.client.batch_post") as post,
+    ):
+        assert client.capture("example_event", distinct_id="user1") is not None
+
+    assert clean_payload.call_count == 2
+    properties = post.call_args.kwargs["batch"][0]["properties"]
+    assert properties["unsupported"] is None
+    assert properties["scrubbed"] is True
+
+
+def test_filter_uses_each_callers_context_without_disabling_the_shared_client():
+    private_context = ContextVar("private_context", default=False)
+    client = Client(
+        FAKE_TEST_API_KEY, should_capture=lambda: not private_context.get(), send=False
+    )
+
+    def capture_in_context(private):
+        token = private_context.set(private)
+        try:
+            return client.capture("example_event", distinct_id="user1")
+        finally:
+            private_context.reset(token)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        results = list(executor.map(capture_in_context, [True, False, True, False]))
+
+    assert results[0] is None
+    assert results[1] is not None
+    assert results[2] is None
+    assert results[3] is not None
+    assert client.disabled is False
+
+
+@pytest.mark.parametrize(
+    "access_flag",
+    [
+        lambda client: client.get_feature_flag_result(
+            "beta-feature", "user1", only_evaluate_locally=True
+        ),
+        lambda client: client.evaluate_flags(
+            "user1", only_evaluate_locally=True
+        ).is_enabled("beta-feature"),
+    ],
+    ids=["get_feature_flag_result", "evaluate_flags"],
+)
+def test_rejected_flag_exposure_is_still_sent_from_an_allowed_context(access_flag):
+    private_context = ContextVar("private_context", default=False)
+    sent_events = []
+
+    def before_send(event):
+        sent_events.append(event["event"])
+        return event
+
+    client = Client(
+        FAKE_TEST_API_KEY,
+        secret_key="test",
+        should_capture=lambda: not private_context.get(),
+        before_send=before_send,
+        send=False,
+    )
+    client.feature_flags = [
+        {
+            "id": 1,
+            "key": "beta-feature",
+            "active": True,
+            "filters": {"groups": [{"properties": [], "rollout_percentage": 100}]},
+        }
+    ]
+
+    token = private_context.set(True)
+    try:
+        access_flag(client)
+    finally:
+        private_context.reset(token)
+    assert sent_events == []
+
+    access_flag(client)
+    access_flag(client)
+    assert sent_events == ["$feature_flag_called"]
+
+
+def test_filter_survives_client_reinitialization_after_fork():
+    should_capture = mock.Mock(return_value=False)
+    client = Client(FAKE_TEST_API_KEY, should_capture=should_capture, send=False)
+
+    client._reinit_after_fork()
+
+    assert client.capture("example_event", distinct_id="user1") is None
+    should_capture.assert_called_once_with()
+
+
+def test_module_setting_applies_at_setup_and_after_initialization():
+    importlib.reload(posthog)
+    try:
+        posthog.api_key = FAKE_TEST_API_KEY
+        posthog.sync_mode = True
+        posthog.should_capture = lambda: False
+
+        with mock.patch("posthog.client.batch_post") as post:
+            assert posthog.capture("rejected", distinct_id="user1") is None
+            posthog.should_capture = lambda: True
+            assert posthog.capture_ai("allowed", distinct_id="user1") is not None
+            posthog.should_capture = None
+            assert posthog.capture("default", distinct_id="user1") is not None
+
+        assert [call.kwargs["batch"][0]["event"] for call in post.call_args_list] == [
+            "allowed",
+            "default",
+        ]
+    finally:
+        if posthog.default_client:
+            posthog.shutdown()
+        importlib.reload(posthog)

references/public_api_snapshot.txt

@@ -642,6 +642,7 @@ attribute posthog.client.Client.queue: Queue
 attribute posthog.client.Client.raw_host = normalize_host(host)
 attribute posthog.client.Client.secret_key = (resolved_secret_key.strip() if isinstance(resolved_secret_key, str) else resolved_secret_key) or None
 attribute posthog.client.Client.send = send
+attribute posthog.client.Client.should_capture = should_capture
 attribute posthog.client.Client.super_properties = super_properties
 attribute posthog.client.Client.sync_mode = sync_mode
 attribute posthog.client.Client.timeout = timeout
@@ -911,6 +912,7 @@ attribute posthog.request.USER_AGENT = 'posthog-python/' + VERSION
 attribute posthog.request.US_INGESTION_ENDPOINT = 'https://us.i.posthog.com'
 attribute posthog.secret_key = None
 attribute posthog.send = True
+attribute posthog.should_capture: Optional[Callable[[], bool]] = None
 attribute posthog.super_properties = None
 attribute posthog.sync_mode = False
 attribute posthog.traces = None
@@ -1042,7 +1044,7 @@ 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, capture_compression: Optional[Union[CaptureCompression, str]] = None, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, traces: Optional[dict] = 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, secret_key=None, metrics: Optional[dict] = None, enable_full_ai_capture=False, capture_trace_context=False, _use_ai_lane=False, _enable_multimodal_capture=False, traces: Optional[dict] = None, should_capture: Optional[Callable[[], bool]] = None)
 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, endpoint=EVENTS_ENDPOINT, max_msg_size=MAX_MSG_SIZE, 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)