feat: add runtime opt-out for capture

.sampo/changesets/opt-out-capturing.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: minor
+---
+
+Add `opt_out_capturing()`, `opt_in_capturing()` and `is_opted_out()` so capture can be suppressed at runtime. While opted out, events are dropped silently with no network request; feature flag evaluation is unaffected. This is separate from the constructor-time `disabled` kill switch, which still cannot be toggled after construction.

posthog/__init__.py

@@ -1224,6 +1224,62 @@ def load_feature_flags():
     return _proxy("load_feature_flags")
 
 
+def opt_out_capturing() -> None:
+    """
+    Stop capturing events on the global client until ``opt_in_capturing()`` is called.
+
+    While opted out, ``capture()`` and every other event-producing API drop
+    their event silently and make no network request. Feature flag evaluation is
+    unaffected. The state is held in memory, so it does not survive a process
+    restart; persist the user's choice yourself and re-apply it on startup.
+
+    Examples:
+        ```python
+        from posthog import opt_out_capturing
+        opt_out_capturing()
+        ```
+
+    Category:
+        Client management
+    """
+    _proxy("opt_out_capturing")
+
+
+def opt_in_capturing() -> None:
+    """
+    Resume capturing events on the global client after ``opt_out_capturing()``.
+
+    Events dropped while opted out are not recovered.
+
+    Examples:
+        ```python
+        from posthog import opt_in_capturing
+        opt_in_capturing()
+        ```
+
+    Category:
+        Client management
+    """
+    _proxy("opt_in_capturing")
+
+
+def is_opted_out() -> bool:
+    """
+    Whether the global client is currently opted out of capturing.
+
+    Examples:
+        ```python
+        from posthog import is_opted_out
+        if not is_opted_out():
+            ...
+        ```
+
+    Category:
+        Client management
+    """
+    return _proxy("is_opted_out")
+
+
 def flush(timeout_seconds: Optional[float] = 10) -> None:
     """
     Tell the client to flush all queued events.

posthog/client.py

@@ -931,6 +931,9 @@ def __init__(
         ] = None
         self._flag_definition_cache_provider_async_runner_lock = threading.Lock()
         self.disabled = disabled or not self.api_key
+        # Runtime consent gate, separate from the constructor-time `disabled`
+        # kill switch, toggled by opt_out_capturing()/opt_in_capturing().
+        self._opted_out = False
         self.disable_geoip = disable_geoip
         self._metrics_config = metrics
         self._metrics: Optional[PostHogMetrics] = None
@@ -2401,7 +2404,7 @@ def _enqueue(self, msg, disable_geoip, lane=None, property_allowlist=None):
         if lane is None:
             lane = self._analytics_lane
 
-        if self.disabled:
+        if self.disabled or self._opted_out:
             return None
 
         timestamp = msg["timestamp"]
@@ -2713,6 +2716,67 @@ def get_active_span(self) -> Optional[Span]:
         """
         return self._active_span_var.get()
 
+    def opt_out_capturing(self) -> None:
+        """
+        Stop capturing events until ``opt_in_capturing()`` is called.
+
+        Details:
+            While opted out, ``capture()`` and every other event-producing API
+            drop their event silently and make no network request. Feature flag
+            evaluation and other non-capture APIs are unaffected. The state is
+            held in memory on this client only, so it does not survive a process
+            restart; persist the user's choice yourself and re-apply it on
+            startup. This is separate from the constructor-time ``disabled``
+            kill switch, which cannot be toggled at runtime.
+
+        Examples:
+            ```python
+            posthog.opt_out_capturing()
+            ```
+
+        Category:
+            Client management
+        """
+        self._opted_out = True
+
+    def opt_in_capturing(self) -> None:
+        """
+        Resume capturing events after ``opt_out_capturing()``.
+
+        Details:
+            Events dropped while opted out are not recovered. A client that was
+            constructed with ``disabled=True`` stays disabled.
+
+        Examples:
+            ```python
+            posthog.opt_in_capturing()
+            ```
+
+        Category:
+            Client management
+        """
+        self._opted_out = False
+
+    def is_opted_out(self) -> bool:
+        """
+        Whether capture is currently suppressed by ``opt_out_capturing()``.
+
+        Returns:
+            ``True`` when the client is opted out. This reports the runtime
+            consent gate only; a client disabled at construction returns
+            ``False`` unless it was also opted out.
+
+        Examples:
+            ```python
+            if not posthog.is_opted_out():
+                posthog.capture("page_viewed", distinct_id="user_123")
+            ```
+
+        Category:
+            Client management
+        """
+        return self._opted_out
+
     def flush(self, timeout_seconds: Optional[float] = 10) -> None:
         """
         Force a flush from the internal queue to the server. Do not use directly, call `shutdown()` instead.

posthog/test/test_client.py

@@ -3587,6 +3587,56 @@ def test_disabled(self):
         self.assertIsNone(msg_uuid)
         self.assertFalse(self.failed)
 
+    def test_opt_out_capturing_drops_events(self):
+        client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, sync_mode=True)
+        self.assertFalse(client.is_opted_out())
+
+        client.opt_out_capturing()
+        self.assertTrue(client.is_opted_out())
+        with mock.patch("posthog.client.batch_post") as patch_post:
+            self.assertIsNone(
+                client.capture("ignored event", distinct_id="distinct_id")
+            )
+            self.assertIsNone(
+                client.set(distinct_id="distinct_id", properties={"plan": "pro"})
+            )
+            self.assertIsNone(client.alias("distinct_id", "other_id"))
+            self.assertIsNone(
+                client.group_identify("company", "id:5", {"name": "PostHog"})
+            )
+            patch_post.assert_not_called()
+        self.assertFalse(self.failed)
+
+    def test_opt_in_capturing_resumes_events(self):
+        client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail)
+        client.opt_out_capturing()
+        self.assertIsNone(client.capture("ignored event", distinct_id="distinct_id"))
+
+        client.opt_in_capturing()
+        self.assertFalse(client.is_opted_out())
+        msg_uuid = client.capture("python test event", distinct_id="distinct_id")
+        self.assertIsNotNone(msg_uuid)
+        client.shutdown()
+        self.assertFalse(self.failed)
+
+    def test_opt_in_capturing_does_not_re_enable_a_disabled_client(self):
+        client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disabled=True)
+        client.opt_in_capturing()
+        self.assertIsNone(
+            client.capture("python test event", distinct_id="distinct_id")
+        )
+        self.assertFalse(self.failed)
+
+    @mock.patch("posthog.client.flags")
+    def test_opt_out_capturing_leaves_feature_flags_working(self, patch_flags):
+        patch_flags.return_value = {"featureFlags": {"beta-feature": True}}
+        client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail)
+        client.opt_out_capturing()
+
+        self.assertTrue(client.feature_enabled("beta-feature", "12345"))
+        patch_flags.assert_called()
+        self.assertFalse(self.failed)
+
     @mock.patch("posthog.client.flags")
     def test_disabled_with_feature_flags(self, patch_flags):
         client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disabled=True)

posthog/test/test_module.py

@@ -48,6 +48,20 @@ def test_alias(self):
         self.assertEqual(event["properties"]["alias"], "distinct_id")
         self.assertEqual(event["uuid"], res)
 
+    def test_opt_out_capturing_drops_module_level_capture(self):
+        with mock.patch.object(posthog, "default_client", self.posthog):
+            posthog.opt_out_capturing()
+            self.assertTrue(posthog.is_opted_out())
+            self.assertIsNone(
+                posthog.capture("ignored event", distinct_id="distinct_id")
+            )
+
+            posthog.opt_in_capturing()
+            self.assertFalse(posthog.is_opted_out())
+            self._assert_enqueue_result(
+                posthog.capture("python module event", distinct_id="distinct_id")
+            )
+
     def test_flush(self):
         self.posthog.flush()
 

references/public_api_snapshot.txt

@@ -1421,6 +1421,7 @@ function posthog.identify_context(distinct_id: str)
 function posthog.integrations.django.markcoroutinefunction(func)
 function posthog.integrations.drf.create_exception_handler(handler: Optional[Callable[[Exception, Mapping[str, Any]], Any]] = None, *, client: Optional[Client] = None, capture_exceptions: Optional[bool] = None, capture_4xx: bool = False, exception_filter: Optional[Callable[[Exception, Any, Mapping[str, Any]], bool]] = None) -> Callable[[Exception, Mapping[str, Any]], Any]
 function posthog.integrations.drf.exception_handler(exc: Exception, context: Mapping[str, Any]) -> Any
+function posthog.is_opted_out() -> bool
 function posthog.join() -> None
 function posthog.load_feature_flags()
 function posthog.mcp.asgi.autowire_stateless_mint(server: Any) -> None
@@ -1437,6 +1438,8 @@ function posthog.mcp.session_token.encode_session_id(payload: SessionTokenPayloa
 function posthog.mcp.session_token.read_mcp_session_header(headers: Any) -> Optional[str]
 function posthog.mcp.tools.get_more_tools_result() -> Dict[str, Any]
 function posthog.new_context(fresh: bool = False, capture_exceptions: Optional[bool] = None, client: Optional[Client] = None)
+function posthog.opt_in_capturing() -> None
+function posthog.opt_out_capturing() -> None
 function posthog.request.batch_post(api_key: str, host: Optional[str] = None, gzip: bool = False, timeout: int = 15, path: str = EVENTS_ENDPOINT, **kwargs) -> requests.Response
 function posthog.request.determine_server_host(host: Optional[str]) -> str
 function posthog.request.disable_connection_reuse() -> None
@@ -1622,9 +1625,12 @@ method posthog.client.Client.get_remote_config_payload(key: str)
 method posthog.client.Client.get_tags() -> Dict[str, Any]
 method posthog.client.Client.group_identify(group_type: str, group_key: str, properties: Optional[Dict[str, Any]] = None, timestamp: Optional[Union[datetime, str]] = None, uuid: Optional[Union[str, UUID]] = None, disable_geoip: Optional[bool] = None, distinct_id: Optional[ID_TYPES] = None) -> Optional[str]
 method posthog.client.Client.identify_context(distinct_id: str) -> None
+method posthog.client.Client.is_opted_out() -> bool
 method posthog.client.Client.join() -> None
 method posthog.client.Client.load_feature_flags()
 method posthog.client.Client.new_context(fresh=False, capture_exceptions: Optional[bool] = None)
+method posthog.client.Client.opt_in_capturing() -> None
+method posthog.client.Client.opt_out_capturing() -> None
 method posthog.client.Client.scoped(fresh=False, capture_exceptions: Optional[bool] = None)
 method posthog.client.Client.set(**kwargs: Unpack[OptionalSetArgs]) -> Optional[str]
 method posthog.client.Client.set_context_device_id(device_id: str) -> None