feat: add analytics_url to send flag analytics to a different host (#215)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: adamvialpando

flagsmith/analytics.py

@@ -31,7 +31,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
@@ -41,8 +45,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

@@ -70,6 +70,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,
@@ -92,6 +93,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
@@ -172,14 +178,12 @@ def __init__(
             self.session.proxies.update(proxies or {})
             retries = retries or Retry(total=3, backoff_factor=0.1)
 
-            api_url = api_url or DEFAULT_API_URL
-            self.api_url = api_url if api_url.endswith("/") else f"{api_url}/"
-
-            realtime_api_url = realtime_api_url or DEFAULT_REALTIME_API_URL
-            self.realtime_api_url = (
-                realtime_api_url
-                if realtime_api_url.endswith("/")
-                else f"{realtime_api_url}/"
+            self.api_url = self._ensure_trailing_slash(api_url or DEFAULT_API_URL)
+            self.realtime_api_url = self._ensure_trailing_slash(
+                realtime_api_url or DEFAULT_REALTIME_API_URL
+            )
+            self.analytics_url = (
+                self._ensure_trailing_slash(analytics_url) if analytics_url else None
             )
 
             self.request_timeout_seconds = request_timeout_seconds
@@ -201,21 +205,30 @@ def __init__(
             self._initialise_analytics(
                 environment_key=environment_key,
                 enable_analytics=enable_analytics,
+                analytics_url=self.analytics_url,
             )
             self._initialise_events(
                 environment_key=environment_key,
                 enable_events=enable_events,
                 event_processor_config=event_processor_config,
             )
 
+    @staticmethod
+    def _ensure_trailing_slash(url: str) -> str:
+        return url if url.endswith("/") else f"{url}/"
+
     def _initialise_analytics(
         self,
         environment_key: str,
         enable_analytics: bool,
+        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,
             )
 
     def _initialise_events(

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,22 @@ 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
+    processor = AnalyticsProcessor(
+        environment_key="test_key",
+        base_api_url="http://edge-proxy/",
+        analytics_url="http://core-api/analytics/flags/",
+    )
+
+    # When
+    with mock.patch("flagsmith.analytics.session") as session:
+        processor.track_feature("my_feature")
+        processor.flush()
+
+    # Then
+    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

@@ -1186,6 +1186,80 @@ def test_get_experiment_flag_falls_back_to_value_without_variant(
     )
 
 
+@responses.activate()
+def test_flagsmith_posts_analytics_to_analytics_url_when_set(
+    api_key: str, flags_json: str, mocker: MockerFixture
+) -> None:
+    # Given
+    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
+    flags = flagsmith.get_environment_flags()
+    assert flags.is_feature_enabled("some_feature") is True
+    assert flagsmith._analytics_processor is not None
+    flagsmith._analytics_processor.flush()
+
+    # Then
+    analytics_posts = [
+        call
+        for call in responses.calls
+        if call.request.method == "POST" and call.request.url == expected_analytics_url
+    ]
+    assert len(analytics_posts) == 1
+    request = analytics_posts[0].request
+    assert request.body is not None
+    assert json.loads(request.body) == {"some_feature": 1}
+    assert request.headers["X-Environment-Key"] == api_key
+    assert not [
+        call
+        for call in responses.calls
+        if call.request.method == "POST"
+        and call.request.url
+        and "edge-proxy" in call.request.url
+    ]
+
+
+@responses.activate()
+def test_flagsmith_posts_analytics_to_api_url_when_analytics_url_unset(
+    api_key: str, flags_json: str, mocker: MockerFixture
+) -> None:
+    # Given
+    mocker.patch("flagsmith.analytics.session", requests.Session())
+    flagsmith = Flagsmith(
+        environment_key=api_key,
+        api_url="http://core-api.flagsmith.com/api/v1/",
+        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
+    flags = flagsmith.get_environment_flags()
+    assert flags.is_feature_enabled("some_feature") is True
+    assert flagsmith._analytics_processor is not None
+    flagsmith._analytics_processor.flush()
+
+    # Then
+    analytics_posts = [
+        call
+        for call in responses.calls
+        if call.request.method == "POST" and call.request.url == expected_analytics_url
+    ]
+    assert len(analytics_posts) == 1
+
+
 def test_get_experiment_flag_skips_exposure_for_disabled_feature(
     mocker: MockerFixture, api_key: str
 ) -> None: