chore: configurabel endpoint for flags local evaluation (#483)

* chore: configurabel endpoint for flags local evaluation

* lint

* greptile suggestions

Author: patricio-posthog

.sampo/changesets/configurable-local-eval-endpoint.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+feat(flags): make local evaluation endpoint configurable via `POSTHOG_LOCAL_EVALUATION_ENDPOINT` env var with fallback to default endpoint

posthog/client.py

@@ -1284,19 +1284,55 @@ def _load_feature_flags(self):
         if should_fetch:
             self._fetch_feature_flags_from_api()
 
+    # Default (Django) endpoint for local evaluation
+    _DEFAULT_LOCAL_EVAL_ENDPOINT = "/api/feature_flag/local_evaluation/"
+
+    def _get_local_eval_endpoint(self):
+        """Get the local evaluation endpoint URL, configurable via env var."""
+        return os.environ.get(
+            "POSTHOG_LOCAL_EVALUATION_ENDPOINT",
+            self._DEFAULT_LOCAL_EVAL_ENDPOINT,
+        )
+
     def _fetch_feature_flags_from_api(self):
         """Fetch feature flags from the PostHog API."""
         try:
             # Store old flags to detect changes
             old_flags_by_key: dict[str, dict] = self.feature_flags_by_key or {}
 
-            response = get(
-                self.personal_api_key,
-                f"/api/feature_flag/local_evaluation/?token={self.api_key}&send_cohorts",
-                self.host,
-                timeout=10,
-                etag=self._flags_etag,
-            )
+            endpoint = self._get_local_eval_endpoint()
+            url = f"{endpoint}?token={self.api_key}&send_cohorts"
+            # Ensure URL has leading slash
+            if not url.startswith("/"):
+                url = f"/{url}"
+
+            try:
+                response = get(
+                    self.personal_api_key,
+                    url,
+                    self.host,
+                    timeout=10,
+                    etag=self._flags_etag,
+                )
+            except Exception as e:
+                # Fall back to the stable Django endpoint when the custom endpoint
+                # (e.g. the Rust-backed /flags/definitions) fails. This enables a
+                # zero-downtime gradual migration: the custom endpoint is tried first
+                # and, on any error, flag evaluation degrades transparently to the
+                # default rather than being blocked entirely.
+                if endpoint != self._DEFAULT_LOCAL_EVAL_ENDPOINT:
+                    self.log.warning(
+                        f"[FEATURE FLAGS] Custom endpoint {endpoint} failed ({e}), falling back to {self._DEFAULT_LOCAL_EVAL_ENDPOINT}"
+                    )
+                    response = get(
+                        self.personal_api_key,
+                        f"{self._DEFAULT_LOCAL_EVAL_ENDPOINT}?token={self.api_key}&send_cohorts",
+                        self.host,
+                        timeout=10,
+                        etag=self._flags_etag,
+                    )
+                else:
+                    raise
 
             # Update stored ETag (clear if server stops sending one)
             self._flags_etag = response.etag

posthog/test/test_feature_flags.py

@@ -1,9 +1,11 @@
 import datetime
+import os
 import unittest
 
 import mock
 from dateutil import parser, tz
 from freezegun import freeze_time
+from parameterized import parameterized
 
 from posthog.client import Client
 from posthog.feature_flags import (
@@ -3780,6 +3782,80 @@ def test_get_all_flags_fallback_when_device_id_missing_for_some_flags(
         self.assertEqual(patch_flags.call_count, 1)
 
 
+class TestLocalEvalEndpointConfig(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.capture_patch = mock.patch.object(Client, "capture")
+        cls.capture_patch.start()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.capture_patch.stop()
+
+    @parameterized.expand(
+        [
+            ("custom_endpoint", "/flags/definitions", "/flags/definitions?"),
+            ("default_endpoint", None, "/api/feature_flag/local_evaluation/"),
+        ]
+    )
+    @mock.patch("posthog.client.get")
+    def test_endpoint_selection(self, _name, env_value, expected_prefix, patch_get):
+        patch_get.return_value = GetResponse(
+            data={"flags": [], "group_type_mapping": {}},
+            etag=None,
+            not_modified=False,
+        )
+        env = {"POSTHOG_LOCAL_EVALUATION_ENDPOINT": env_value} if env_value else {}
+        with mock.patch.dict("os.environ", env, clear=False):
+            if env_value is None:
+                os.environ.pop("POSTHOG_LOCAL_EVALUATION_ENDPOINT", None)
+            client = Client(FAKE_TEST_API_KEY, personal_api_key="test-key")
+            client._fetch_feature_flags_from_api()
+            call_url = patch_get.call_args[0][1]
+            self.assertTrue(
+                call_url.startswith(expected_prefix),
+                f"Expected URL starting with {expected_prefix}, got: {call_url}",
+            )
+
+    @parameterized.expand(
+        [
+            ("custom_endpoint_falls_back", "/flags/definitions", 2),
+            ("default_endpoint_no_fallback", None, 1),
+        ]
+    )
+    @mock.patch("posthog.client.get")
+    def test_endpoint_fallback_on_failure(
+        self, _name, env_value, expected_call_count, patch_get
+    ):
+        success_response = GetResponse(
+            data={"flags": [], "group_type_mapping": {}},
+            etag=None,
+            not_modified=False,
+        )
+        if expected_call_count == 2:
+            patch_get.side_effect = [Exception("connection refused"), success_response]
+        else:
+            patch_get.side_effect = Exception("connection refused")
+
+        env = {"POSTHOG_LOCAL_EVALUATION_ENDPOINT": env_value} if env_value else {}
+        with mock.patch.dict("os.environ", env, clear=False):
+            if env_value is None:
+                os.environ.pop("POSTHOG_LOCAL_EVALUATION_ENDPOINT", None)
+            client = Client(FAKE_TEST_API_KEY, personal_api_key="test-key")
+            client._fetch_feature_flags_from_api()
+            self.assertEqual(patch_get.call_count, expected_call_count)
+            if expected_call_count == 2:
+                # First call used custom endpoint, second fell back to default
+                self.assertTrue(
+                    patch_get.call_args_list[0][0][1].startswith("/flags/definitions?")
+                )
+                self.assertTrue(
+                    patch_get.call_args_list[1][0][1].startswith(
+                        "/api/feature_flag/local_evaluation/"
+                    )
+                )
+
+
 class TestMatchProperties(unittest.TestCase):
     def property(self, key, value, operator=None):
         result = {"key": key, "value": value}