feat: add analytics_url to send flag analytics to a different host

README.md

@@ -10,6 +10,25 @@ The SDK for Python applications for [https://www.flagsmith.com/](https://www.fla
 For full documentation visit
 [https://docs.flagsmith.com/clients/server-side?language=python](https://docs.flagsmith.com/clients/server-side?language=python).
 
+### Sending flag analytics to a different host than evaluations
+
+When evaluating flags through an Edge Proxy (or another host that does not handle the analytics endpoint), pass
+`analytics_url` to send flag analytics directly to the core Flagsmith API while keeping flag evaluations on the proxy:
+
+```python
+from flagsmith import Flagsmith
+
+flagsmith = Flagsmith(
+    environment_key="<your API key>",
+    api_url="https://edge-proxy.internal/api/v1/",
+    analytics_url="https://edge.api.flagsmith.com/api/v1/analytics/flags/",
+    enable_local_evaluation=True,
+    enable_analytics=True,
+)
+```
+
+When `analytics_url` is unset, analytics continue to post to `<api_url>/analytics/flags/` as before.
+
 ## Contributing
 
 Please read [CONTRIBUTING.md](CONTRIBUTING.md) for details on our code of conduct, and the process for submitting pull

flagsmith/analytics.py

@@ -28,7 +28,11 @@ class AnalyticsProcessor:
     """
 
     def __init__(
-        self, environment_key: str, base_api_url: str, timeout: typing.Optional[int] = 3
+        self,
+        environment_key: str,
+        base_api_url: str,
+        timeout: typing.Optional[int] = 3,
+        analytics_url: typing.Optional[str] = None,
     ):
         """
         Initialise the AnalyticsProcessor to handle sending analytics on flag usage to
@@ -38,8 +42,13 @@ def __init__(
         :param base_api_url: base api url to override when using self hosted version
         :param timeout: used to tell requests to stop waiting for a response after a
             given number of seconds
+        :param analytics_url: full URL of the flag analytics endpoint, used to override
+            the default ``<base_api_url>/analytics/flags/``. Intended for deployments
+            where flag evaluation traffic and analytics traffic must go to different
+            hosts (for example, evaluating through the Edge Proxy while sending
+            analytics to the core API).
         """
-        self.analytics_endpoint = base_api_url + ANALYTICS_ENDPOINT
+        self.analytics_endpoint = analytics_url or (base_api_url + ANALYTICS_ENDPOINT)
         self.environment_key = environment_key
         self._last_flushed = datetime.now()
         self.analytics_data: typing.MutableMapping[str, typing.Any] = {}

flagsmith/flagsmith.py

@@ -68,6 +68,7 @@ def __init__(
         environment_key: typing.Optional[str] = None,
         api_url: typing.Optional[str] = None,
         realtime_api_url: typing.Optional[str] = None,
+        analytics_url: typing.Optional[str] = None,
         custom_headers: typing.Optional[typing.Dict[str, typing.Any]] = None,
         request_timeout_seconds: typing.Optional[int] = 10,
         enable_local_evaluation: bool = False,
@@ -89,6 +90,11 @@ def __init__(
             Required unless offline_mode is True.
         :param api_url: Override the URL of the Flagsmith API to communicate with
         :param realtime_api_url: Override the URL of the Flagsmith real-time API
+        :param analytics_url: Override the URL used for flag analytics requests when
+            enable_analytics is True. When unset, analytics are posted to
+            ``<api_url>/analytics/flags/``. Set this when api_url points at a host that
+            does not handle analytics (for example, the Edge Proxy) so analytics can be
+            sent directly to the core Flagsmith API.
         :param custom_headers: Additional headers to add to requests made to the
             Flagsmith API
         :param request_timeout_seconds: Number of seconds to wait for a request to
@@ -170,6 +176,10 @@ def __init__(
                 else f"{realtime_api_url}/"
             )
 
+            if analytics_url and not analytics_url.endswith("/"):
+                analytics_url = f"{analytics_url}/"
+            self.analytics_url = analytics_url
+
             self.request_timeout_seconds = request_timeout_seconds
             self.session.mount(self.api_url, HTTPAdapter(max_retries=retries))
 
@@ -190,17 +200,22 @@ def __init__(
                 environment_key=environment_key,
                 enable_analytics=enable_analytics,
                 pipeline_analytics_config=pipeline_analytics_config,
+                analytics_url=self.analytics_url,
             )
 
     def _initialise_analytics(
         self,
         environment_key: str,
         enable_analytics: bool,
         pipeline_analytics_config: typing.Optional[PipelineAnalyticsConfig],
+        analytics_url: typing.Optional[str] = None,
     ) -> None:
         if enable_analytics:
             self._analytics_processor = AnalyticsProcessor(
-                environment_key, self.api_url, timeout=self.request_timeout_seconds
+                environment_key,
+                self.api_url,
+                timeout=self.request_timeout_seconds,
+                analytics_url=analytics_url,
             )
         if pipeline_analytics_config:
             self._pipeline_analytics_processor = PipelineAnalyticsProcessor(

tests/test_analytics.py

@@ -2,7 +2,11 @@
 from datetime import datetime, timedelta
 from unittest import mock
 
-from flagsmith.analytics import ANALYTICS_TIMER, AnalyticsProcessor
+from flagsmith.analytics import (
+    ANALYTICS_ENDPOINT,
+    ANALYTICS_TIMER,
+    AnalyticsProcessor,
+)
 
 
 def test_analytics_processor_track_feature_updates_analytics_data(
@@ -36,6 +40,9 @@ def test_analytics_processor_flush_post_request_data_match_ananlytics_data(
     # Then
     session.post.assert_called()
     post_call = session.mock_calls[0]
+    # When analytics_url is unset, the POST falls back to base_api_url + ANALYTICS_ENDPOINT.
+    # Locks the default in so a future refactor cannot silently break it.
+    assert post_call[1][0] == "http://test_url" + ANALYTICS_ENDPOINT
     assert {"my_feature_1": 1, "my_feature_2": 1} == json.loads(post_call[2]["data"])
 
 
@@ -66,3 +73,24 @@ def test_analytics_processor_calling_track_feature_calls_flush_when_timer_runs_o
 
     # Then
     session.post.assert_called()
+
+
+def test_analytics_processor_posts_to_analytics_url_when_set() -> None:
+    # Given an AnalyticsProcessor configured to send analytics to a host
+    # that is different from base_api_url (e.g. base_api_url points at an
+    # Edge Proxy that does not handle analytics)
+    processor = AnalyticsProcessor(
+        environment_key="test_key",
+        base_api_url="http://edge-proxy/",
+        analytics_url="http://core-api/analytics/flags/",
+    )
+
+    # When the processor flushes
+    with mock.patch("flagsmith.analytics.session") as session:
+        processor.track_feature("my_feature")
+        processor.flush()
+
+    # Then the POST goes to analytics_url and never to the edge-proxy host
+    session.post.assert_called_once()
+    assert session.post.call_args[0][0] == "http://core-api/analytics/flags/"
+    assert "edge-proxy" not in session.post.call_args[0][0]

tests/test_flagsmith.py

@@ -1038,3 +1038,52 @@ def test_identity_flags_records_evaluation_with_resolved_traits(
         identity_identifier="user123",
         traits={"plan": "premium"},
     )
+
+
+@responses.activate()
+def test_flagsmith_posts_analytics_to_analytics_url_when_set(
+    api_key: str, flags_json: str, mocker: MockerFixture
+) -> None:
+    # Given a Flagsmith client pointed at an Edge Proxy for flag evaluations,
+    # with analytics_url overriding the analytics endpoint to the core API.
+    # analytics_url is intentionally written without a trailing slash to
+    # exercise the constructor's normalisation.
+    #
+    # We swap the fire-and-forget FuturesSession for a plain requests.Session
+    # so the analytics POST happens synchronously on the test thread and is
+    # observable via responses.calls; without this swap the worker thread can
+    # race the assertions.
+    mocker.patch("flagsmith.analytics.session", requests.Session())
+    flagsmith = Flagsmith(
+        environment_key=api_key,
+        api_url="http://edge-proxy.internal/api/v1/",
+        analytics_url="http://core-api.flagsmith.com/api/v1/analytics/flags",
+        enable_analytics=True,
+    )
+
+    expected_analytics_url = "http://core-api.flagsmith.com/api/v1/analytics/flags/"
+    responses.add(method="GET", url=flagsmith.environment_flags_url, body=flags_json)
+    responses.add(method="POST", url=expected_analytics_url, status=200)
+
+    # When the customer-facing evaluation API is exercised. This is the path
+    # that triggers track_feature internally (Flags.get_flag in models.py).
+    flags = flagsmith.get_environment_flags()
+    assert flags.is_feature_enabled("some_feature") is True
+
+    # Force the flush deterministically rather than waiting on ANALYTICS_TIMER.
+    assert flagsmith._analytics_processor is not None
+    flagsmith._analytics_processor.flush()
+
+    # Then exactly one analytics POST landed on the override host (with the
+    # trailing slash applied), carried the tracked feature payload, and the
+    # Edge Proxy never received an analytics request.
+    analytics_calls = [
+        call for call in responses.calls if call.request.method == "POST"
+    ]
+    assert len(analytics_calls) == 1
+    request = analytics_calls[0].request
+    assert request.url == expected_analytics_url
+    assert "edge-proxy" not in request.url
+    assert request.body is not None
+    assert json.loads(request.body) == {"some_feature": 1}
+    assert request.headers["X-Environment-Key"] == api_key