Update exposure event tracking and add docs

efahk · Kwame Efah · commit 1cbc9c749540 · 2025-09-18T15:33:43.000-07:00
diff --git a/demo/local_flags.py b/demo/local_flags.py
@@ -6,7 +6,7 @@
 logging.basicConfig(level=logging.INFO)
 
 # Configure your project token, the feature flag  to test, and user context to evaluate.
-PROJECT_TOKEN = ""
+PROJECT_TOKEN = "044781e247eabd8e9b73fad8a8c093d2"
 FLAG_KEY = "sample-flag"
 FLAG_FALLBACK_VARIANT = "control"
 USER_CONTEXT = { "distinct_id": "sample-distinct-id" }
@@ -18,14 +18,14 @@
 # Use the correct data residency endpoint for your project.
 API_HOST = "api-eu.mixpanel.com"
 
-async def main():
+def main():
     local_config = mixpanel.LocalFlagsConfig(api_host=API_HOST, enable_polling=SHOULD_POLL_CONTINOUSLY, polling_interval_in_seconds=POLLING_INTERVAL_IN_SECONDS)
 
     # Optionally use mixpanel client as a context manager, that will ensure shutdown of resources used by feature flagging
-    async with mixpanel.Mixpanel(PROJECT_TOKEN, local_flags_config=local_config) as mp:
-        await mp.local_flags.astart_polling_for_definitions()
+    with mixpanel.Mixpanel(PROJECT_TOKEN, local_flags_config=local_config) as mp:
+        mp.local_flags.start_polling_for_definitions()
         variant_value = mp.local_flags.get_variant_value(FLAG_KEY, FLAG_FALLBACK_VARIANT, USER_CONTEXT)
         print(f"Variant value: {variant_value}")
 
 if __name__ == '__main__':
-    asyncio.run(main())
+    main()
diff --git a/mixpanel/flags/local_feature_flags.py b/mixpanel/flags/local_feature_flags.py
@@ -5,7 +5,6 @@
 import threading
 from datetime import datetime, timedelta
 from typing import Dict, Any, Callable, Optional
-from concurrent.futures import Future, ThreadPoolExecutor
 from .types import ExperimentationFlag, ExperimentationFlags, SelectedVariant, LocalFlagsConfig, Rollout
 from .utils import REQUEST_HEADERS, normalized_hash, prepare_common_query_params, EXPOSURE_EVENT
 
@@ -16,13 +15,20 @@ class LocalFeatureFlagsProvider:
     FLAGS_DEFINITIONS_URL_PATH = "/flags/definitions"
 
     def __init__(self, token: str, config: LocalFlagsConfig, version: str, tracker: Callable) -> None:
+        """
+        Initializes the LocalFeatureFlagsProvider
+        :param str token: your project's Mixpanel token
+        :param LocalFlagsConfig config: configuration options for the local feature flags provider
+        :param str version: the version of the Mixpanel library being used, just for tracking
+        :param str tracker: A function used to track flags exposure events to mixpanel 
+        """
         self._token: str = token
         self._config: LocalFlagsConfig = config
         self._version = version
         self._tracker: Callable = tracker
-        self._executor: ThreadPoolExecutor = config.custom_executor or ThreadPoolExecutor(max_workers=5)
 
         self._flag_definitions: Dict[str, ExperimentationFlag] = dict()
+        self._are_flags_ready = False
 
         httpx_client_parameters = {
             "base_url": f"https://{config.api_host}",
@@ -37,29 +43,41 @@ def __init__(self, token: str, config: LocalFlagsConfig, version: str, tracker:
         self._sync_client: httpx.Client = httpx.Client(**httpx_client_parameters)
 
         self._async_polling_task: Optional[asyncio.Task] = None
-        self._sync_polling_task: Optional[Future] = None
+        self._sync_polling_task: Optional[threading.Thread] = None
 
         self._sync_stop_event = threading.Event()
 
     def start_polling_for_definitions(self):
+        """
+        Fetches flag definitions for the current project.
+        If configured by the caller, starts a background thread to poll for updates at regular intervals, if one does not already exist. 
+        """
         self._fetch_flag_definitions()
 
         if self._config.enable_polling:
             if not self._sync_polling_task and not self._async_polling_task:
                 self._sync_stop_event.clear()
-                self._sync_polling_task = self._executor.submit(self._start_continuous_polling)
+                self._sync_polling_task = threading.Thread(target=self._start_continuous_polling, daemon=True)
+                self._sync_polling_task.start()
             else:
-                logging.error("A polling task is already running")
+                logging.warning("A polling task is already running")
 
     def stop_polling_for_definitions(self):
+        """
+        If there exists a reference to a background thread polling for flag definition updates, signal it to stop and clear the reference.
+        Once stopped, the polling thread cannot be restarted.
+        """
         if self._sync_polling_task:
             self._sync_stop_event.set()
-            self._sync_polling_task.cancel()
             self._sync_polling_task = None
         else:
             logging.info("There is no polling task to cancel.")
 
     async def astart_polling_for_definitions(self):
+        """
+        Fetches flag definitions for the current project.
+        If configured by the caller, starts an async task on the event loop to poll for updates at regular intervals, if one does not already exist.
+        """
         await self._afetch_flag_definitions()
 
         if self._config.enable_polling:
@@ -69,6 +87,9 @@ async def astart_polling_for_definitions(self):
                 logging.error("A polling task is already running")
 
     async def astop_polling_for_definitions(self):
+        """
+        If there exists an async task  to poll for flag definition updates, cancel the task and clear the reference to it.
+        """
         if self._async_polling_task:
             self._async_polling_task.cancel()
             self._async_polling_task = None
@@ -94,20 +115,39 @@ def _start_continuous_polling(self):
 
     def are_flags_ready(self) -> bool:
         """
-        Check if flag definitions have been loaded and are ready for use.
-        :return: True if flag definitions are populated, False otherwise.
+        Check if the call to fetch flag definitions has been made successfully. 
         """
-        return bool(self._flag_definitions)
+        return self._are_flags_ready
 
     def get_variant_value(self, flag_key: str, fallback_value: Any, context: Dict[str, Any]) -> Any:
+        """
+        Get the value of a feature flag variant.
+
+        :param str flag_key: The key of the feature flag to evaluate
+        :param Any fallback_value: The default value to return if the flag is not found or evaluation fails
+        :param Dict[str, Any] context: Context dictionary containing user's distinct_id and any other attributes needed for rollout evaluation 
+        """
         variant = self.get_variant(flag_key, SelectedVariant(variant_value=fallback_value), context)
         return variant.variant_value
 
     def is_enabled(self, flag_key: str, context: Dict[str, Any]) -> bool:
+        """
+        Check if a feature flag is enabled for the given context.
+
+        :param str flag_key: The key of the feature flag to check
+        :param Dict[str, Any] context: Context dictionary containing user's distinct_id and any other attributes needed for rollout evaluation 
+        """
         variant_value = self.get_variant_value(flag_key, False, context)
         return bool(variant_value)
 
     def get_variant(self, flag_key: str, fallback_value: SelectedVariant, context: Dict[str, Any]) -> SelectedVariant:
+        """
+        Gets the selected variant for a feature flag
+
+        :param str flag_key: The key of the feature flag to evaluate
+        :param SelectedVariant fallback_value: The default variant to return if evaluation fails
+        :param Dict[str, Any] context: Context dictionary containing user's distinct_id and any other attributes needed for rollout evaluation 
+        """
         start_time = time.perf_counter()
         flag_definition = self._flag_definitions.get(flag_key)
 
@@ -125,7 +165,7 @@ def get_variant(self, flag_key: str, fallback_value: SelectedVariant, context: D
         if rollout := self._get_assigned_rollout(flag_definition, context_value, context):
             variant = self._get_assigned_variant(flag_definition, context_value, flag_key, rollout)
             end_time = time.perf_counter()
-            self.track_exposure(flag_key, variant, end_time - start_time, context)
+            self._track_exposure(flag_key, variant, end_time - start_time, context)
             return variant
 
         logger.info(f"{flag_definition.context} context {context_value} not eligible for any rollout for flag: {flag_key}")
@@ -237,10 +277,11 @@ def _handle_response(self, response: httpx.Response, start_time: datetime, end_t
             logger.exception("Failed to parse flag definitions")
 
         self._flag_definitions = flags
+        self._are_flags_ready = True
         logger.info(f"Successfully fetched {len(self._flag_definitions)} flag definitions")
 
 
-    def track_exposure(self, flag_key: str, variant: SelectedVariant, latency_in_seconds: float, context: Dict[str, Any]):
+    def _track_exposure(self, flag_key: str, variant: SelectedVariant, latency_in_seconds: float, context: Dict[str, Any]):
         if distinct_id := context.get("distinct_id"):
             properties = {
                 'Experiment name': flag_key,
@@ -249,7 +290,8 @@ def track_exposure(self, flag_key: str, variant: SelectedVariant, latency_in_sec
                 "Flag evaluation mode": "local",
                 "Variant fetch latency (ms)": latency_in_seconds * 1000
             }
-            self._executor.submit(self._tracker, distinct_id, EXPOSURE_EVENT, properties)
+
+            self._tracker(distinct_id, EXPOSURE_EVENT, properties)
         else:
             logging.error("Cannot track exposure event without a distinct_id in the context")
 
diff --git a/mixpanel/flags/remote_feature_flags.py b/mixpanel/flags/remote_feature_flags.py
@@ -8,7 +8,6 @@
 from asgiref.sync import sync_to_async
 
 from .types import RemoteFlagsConfig, SelectedVariant, RemoteFlagsResponse
-from concurrent.futures import ThreadPoolExecutor
 from .utils import REQUEST_HEADERS, EXPOSURE_EVENT, prepare_common_query_params
 
 logger = logging.getLogger(__name__)
@@ -22,7 +21,6 @@ def __init__(self, token: str, config: RemoteFlagsConfig, version: str, tracker:
         self._config: RemoteFlagsConfig = config
         self._version: str = version
         self._tracker: Callable = tracker
-        self._executor: ThreadPoolExecutor = config.custom_executor or ThreadPoolExecutor(max_workers=5)
 
         httpx_client_parameters = {
             "base_url": f"https://{config.api_host}",
@@ -36,10 +34,24 @@ def __init__(self, token: str, config: RemoteFlagsConfig, version: str, tracker:
         self._request_params_base = prepare_common_query_params(self._token, version)
 
     async def aget_variant_value(self, flag_key: str, fallback_value: Any, context: Dict[str, Any]) -> Any:
+        """
+        Gets the selected variant value of a feature flag variant for the current user context from remote server.
+
+        :param str flag_key: The key of the feature flag to evaluate
+        :param Any fallback_value: The default value to return if the flag is not found or evaluation fails
+        :param Dict[str, Any] context: Context dictionary containing user attributes and rollout context
+        """
         variant = await self.aget_variant(flag_key, SelectedVariant(variant_value=fallback_value), context)
         return variant.variant_value
 
     async def aget_variant(self, flag_key: str, fallback_value: SelectedVariant, context: Dict[str, Any]) -> SelectedVariant:
+        """
+        Asynchronously gets the selected variant  of a feature flag variant for the current user context from remote server.
+
+        :param str flag_key: The key of the feature flag to evaluate
+        :param SelectedVariant fallback_value: The default variant to return if evaluation fails
+        :param Dict[str, Any] context: Context dictionary containing user attributes and rollout context
+        """
         try:
             params = self._prepare_query_params(flag_key, context)
             start_time = datetime.now()
@@ -51,22 +63,42 @@ async def aget_variant(self, flag_key: str, fallback_value: SelectedVariant, con
             if not is_fallback and (distinct_id := context.get("distinct_id")):
                 properties = self._build_tracking_properties(flag_key, selected_variant, start_time, end_time)
                 asyncio.create_task(
-                    sync_to_async(self._tracker, executor=self._executor, thread_sensitive=False)(distinct_id, EXPOSURE_EVENT, properties))
+                    sync_to_async(self._tracker, thread_sensitive=False)(distinct_id, EXPOSURE_EVENT, properties))
 
             return selected_variant
         except Exception:
             logging.exception(f"Failed to get remote variant for flag '{flag_key}'")
             return fallback_value
 
     async def ais_enabled(self, flag_key: str, context: Dict[str, Any]) -> bool:
+        """
+        Asynchronously checks if a feature flag is enabled for the given context.
+
+        :param str flag_key: The key of the feature flag to check
+        :param Dict[str, Any] context: Context dictionary containing user attributes and rollout context
+        """
         variant_value = await self.aget_variant_value(flag_key, False, context)
         return bool(variant_value)
 
     def get_variant_value(self, flag_key: str, fallback_value: Any, context: Dict[str, Any]) -> Any:
+        """
+        Synchronously gets the value of a feature flag variant from remote server.
+
+        :param str flag_key: The key of the feature flag to evaluate
+        :param Any fallback_value: The default value to return if the flag is not found or evaluation fails
+        :param Dict[str, Any] context: Context dictionary containing user attributes and rollout context
+        """
         variant = self.get_variant(flag_key, SelectedVariant(variant_value=fallback_value), context)
         return variant.variant_value
 
     def get_variant(self, flag_key: str, fallback_value: SelectedVariant, context: Dict[str, Any]) -> SelectedVariant:
+        """
+        Synchronously getsthe selected variant for a feature flag from remote server.
+
+        :param str flag_key: The key of the feature flag to evaluate
+        :param SelectedVariant fallback_value: The default variant to return if evaluation fails
+        :param Dict[str, Any] context: Context dictionary containing user attributes and rollout context
+        """
         try:
             params = self._prepare_query_params(flag_key, context)
             start_time = datetime.now()
@@ -77,14 +109,20 @@ def get_variant(self, flag_key: str, fallback_value: SelectedVariant, context: D
 
             if not is_fallback and (distinct_id := context.get("distinct_id")):
                 properties = self._build_tracking_properties(flag_key, selected_variant, start_time, end_time)
-                self._executor.submit(self._tracker, distinct_id, EXPOSURE_EVENT, properties)
+                self._tracker(distinct_id, EXPOSURE_EVENT, properties)
 
             return selected_variant
         except Exception:
             logging.exception(f"Failed to get remote variant for flag '{flag_key}'")
             return fallback_value
 
     def is_enabled(self, flag_key: str, context: Dict[str, Any]) -> bool:
+        """
+        Synchronously checks if a feature flag is enabled for the given context.
+
+        :param str flag_key: The key of the feature flag to check
+        :param Dict[str, Any] context: Context dictionary containing user attributes and rollout context
+        """
         variant_value = self.get_variant_value(flag_key, False, context)
         return bool(variant_value)
 
diff --git a/mixpanel/flags/test_local_feature_flags.py b/mixpanel/flags/test_local_feature_flags.py
@@ -227,22 +227,19 @@ async def test_get_variant_value_tracks_exposure_when_variant_selected(self):
         with patch('mixpanel.flags.utils.normalized_hash') as mock_hash:
             mock_hash.return_value = 0.5
             _ = flags.get_variant_value("test_flag", "fallback", {"distinct_id": "user123"})
-            flags._executor.shutdown()
             flags._tracker.assert_called_once()
 
     @respx.mock
     async def test_get_variant_value_does_not_track_exposure_on_fallback(self):
         flags = await self.setup_flags([])
         _ = flags.get_variant_value("nonexistent_flag", "fallback", {"distinct_id": "user123"})
-        flags._executor.shutdown()
         flags._tracker.assert_not_called()
 
     @respx.mock
     async def test_get_variant_value_does_not_track_exposure_without_distinct_id(self):
         flag = create_test_flag(context="company")
         flags = await self.setup_flags([flag])
         _ = flags.get_variant_value("nonexistent_flag", "fallback", {"company_id": "company123"})
-        flags._executor.shutdown()
         flags._tracker.assert_not_called()
 
     @respx.mock
diff --git a/mixpanel/flags/test_remote_feature_flags.py b/mixpanel/flags/test_remote_feature_flags.py
@@ -64,14 +64,12 @@ async def test_get_variant_value_tracks_exposure_event_if_variant_selected(self)
         if pending:
             await asyncio.gather(*pending, return_exceptions=True)
 
-        self._flags._executor.shutdown()
         self.mock_tracker.assert_called_once()
 
     @respx.mock
     async def test_get_variant_value_does_not_track_exposure_event_if_fallback(self): 
         respx.get(ENDPOINT).mock(side_effect=httpx.RequestError("Network error"))
         await self._flags.aget_variant_value("test_flag", "control", {"distinct_id": "user123"})
-        self._flags._executor.shutdown()
         self.mock_tracker.assert_not_called()
 
     @respx.mock
@@ -135,14 +133,12 @@ def test_get_variant_value_tracks_exposure_event_if_variant_selected(self):
             return_value=create_success_response({"test_flag": SelectedVariant(variant_key="treatment", variant_value="treatment")}))
 
         self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"})
-        self._flags._executor.shutdown()
         self.mock_tracker.assert_called_once()
 
     @respx.mock
     def test_get_variant_value_does_not_track_exposure_event_if_fallback(self):
         respx.get(ENDPOINT).mock(side_effect=httpx.RequestError("Network error"))
         self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"})
-        self._flags._executor.shutdown()
         self.mock_tracker.assert_not_called()
 
     @respx.mock
diff --git a/mixpanel/flags/types.py b/mixpanel/flags/types.py
@@ -1,5 +1,4 @@
 from typing import Optional, List, Dict, Any
-from concurrent.futures import ThreadPoolExecutor
 from pydantic import BaseModel, ConfigDict
 
 MIXPANEL_DEFAULT_API_ENDPOINT = "api.mixpanel.com"
@@ -9,7 +8,6 @@ class FlagsConfig(BaseModel):
 
     api_host: str = "api.mixpanel.com"
     request_timeout_in_seconds: int = 10
-    custom_executor: Optional[ThreadPoolExecutor] = None
 
 class LocalFlagsConfig(FlagsConfig):
     enable_polling: bool = True
diff --git a/uv.lock b/uv.lock
