From 7885897bdc06fe1b391b5e10453c389ed51dd744 Mon Sep 17 00:00:00 2001 From: Kwame Efah <37164746+efahk@users.noreply.github.com> Date: Mon, 25 Aug 2025 17:07:26 +0000 Subject: [PATCH] Implement feature flag support --- CHANGES.txt | 3 + demo/flags.py | 37 ++++ demo/local_flags.py | 35 ++++ demo/remote_flags.py | 38 ++++ mixpanel/__init__.py | 25 ++- mixpanel/flags/__init__.py | 0 mixpanel/flags/local_feature_flags.py | 203 ++++++++++++++++++ mixpanel/flags/remote_feature_flags.py | 120 +++++++++++ mixpanel/flags/test_local_feature_flags.py | 221 ++++++++++++++++++++ mixpanel/flags/test_remote_feature_flags.py | 72 +++++++ mixpanel/flags/types.py | 55 +++++ mixpanel/flags/utils.py | 59 ++++++ pyproject.toml | 9 +- 13 files changed, 873 insertions(+), 4 deletions(-) create mode 100644 demo/flags.py create mode 100644 demo/local_flags.py create mode 100644 demo/remote_flags.py create mode 100644 mixpanel/flags/__init__.py create mode 100644 mixpanel/flags/local_feature_flags.py create mode 100644 mixpanel/flags/remote_feature_flags.py create mode 100644 mixpanel/flags/test_local_feature_flags.py create mode 100644 mixpanel/flags/test_remote_feature_flags.py create mode 100644 mixpanel/flags/types.py create mode 100644 mixpanel/flags/utils.py diff --git a/CHANGES.txt b/CHANGES.txt index bdf842b..9d1b9e3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,6 @@ +v5.0.0-rc1 +* Added feature flagging support + v4.11.0 * Set minimum supported python version to 3.9, deprecating support for end-of-life versions of python * Convert setup.py to pyproject.toml diff --git a/demo/flags.py b/demo/flags.py new file mode 100644 index 0000000..d715a5f --- /dev/null +++ b/demo/flags.py @@ -0,0 +1,37 @@ +#import sys +import os +#sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +import mixpanel + +if __name__ == '__main__': + mp = mixpanel.Mixpanel("391d3916270285cbf9f433f51a99a44c") + + # Example 1: Local flags (with flag definitions fetched and cached locally) + print("=== LOCAL FLAGS EXAMPLE ===") + local_config = mixpanel.LocalFlagsConfig(api_host="api.mixpanel.com") + local_flags_provider = mp.getLocalFlagsProvider(local_config) + + user_context = { + "distinct_id": "test-user-1" + } + + local_variant = local_flags_provider.get_variant("funnel_reentry", "control", user_context) + print("LOCAL VARIANT IS ", local_variant) + + # Example 2: Remote flags (flags evaluated remotely on each call) + print("\n=== REMOTE FLAGS EXAMPLE ===") + remote_config = mixpanel.RemoteFlagsConfig(api_host="api.mixpanel.com") + remote_flags_provider = mp.getRemoteFlagsProvider(remote_config) + + import asyncio + async def test_remote_flags(): + async with remote_flags_provider: + remote_variant = await remote_flags_provider.get_variant("funnel_reentry", "control", user_context) + print("REMOTE VARIANT (async) IS ", remote_variant) + + asyncio.run(test_remote_flags()) + + # Using sync version + remote_variant_sync = remote_flags_provider.get_variant_sync("funnel_reentry", "control", user_context) + print("REMOTE VARIANT (sync) IS ", remote_variant_sync) \ No newline at end of file diff --git a/demo/local_flags.py b/demo/local_flags.py new file mode 100644 index 0000000..78a6271 --- /dev/null +++ b/demo/local_flags.py @@ -0,0 +1,35 @@ +import os +import asyncio +import mixpanel + +async def main(): + mp = mixpanel.Mixpanel("391d3916270285cbf9f433f51a99a44c") + + print("=== LOCAL FLAGS EXAMPLE ===") + + local_config = mixpanel.LocalFlagsConfig() + local_config.api_host = "api.mixpanel.com" + local_config.enablePolling = True + local_config.pollingIntervalInSeconds = 60 + + async with mp.getLocalFlagsProvider(local_config) as local_flags_provider: + await local_flags_provider.start_polling_for_definitions() + + user_context = { + "distinct_id": "test-user-1" + } + + variant_value = local_flags_provider.get_variant_value("test-flag", "control", user_context) + print(f"Variant value: {variant_value}") + + variant = local_flags_provider.get_variant("test-flag", + mixpanel.SelectedVariant(variant_key="control", variant_value="control"), + user_context) + + print(f"Full variant: key={variant.variant_key}, value={variant.variant_value}") + + is_enabled = local_flags_provider.is_enabled("funnel_reentry", user_context) + print(f"Feature enabled: {is_enabled}") + +if __name__ == '__main__': + asyncio.run(main()) \ No newline at end of file diff --git a/demo/remote_flags.py b/demo/remote_flags.py new file mode 100644 index 0000000..b77b071 --- /dev/null +++ b/demo/remote_flags.py @@ -0,0 +1,38 @@ +import asyncio +import mixpanel + +async def main(): + mp = mixpanel.Mixpanel("391d3916270285cbf9f433f51a99a44c") + remote_config = mixpanel.RemoteFlagsConfig() + remote_flags_provider = mp.getRemoteFlagsProvider(remote_config) + user_context = { + "distinct_id": "test-user-1" + } + + print("=== ASYNC VERSION ===") + async with remote_flags_provider: + variant_value = await remote_flags_provider.get_variant_value("sample-flag", "control", user_context) + print(f"Async variant value: {variant_value}") + + variant = await remote_flags_provider.get_variant("sample-flag", + mixpanel.SelectedVariant(variant_key="control", variant_value="control"), + user_context) + print(f"Async full variant: key={variant.variant_key}, value={variant.variant_value}") + + is_enabled = await remote_flags_provider.is_enabled("sample-flag", user_context) + print(f"Async feature enabled: {is_enabled}") + + print("\n=== SYNC VERSION ===") + variant_value_sync = remote_flags_provider.get_variant_value_sync("sample-flag", "control", user_context) + print(f"Sync variant value: {variant_value_sync}") + + variant_sync = remote_flags_provider.get_variant_sync("sample-flag", + mixpanel.SelectedVariant(variant_key="control", variant_value="control"), + user_context) + print(f"Sync full variant: key={variant_sync.variant_key}, value={variant_sync.variant_value}") + + is_enabled_sync = remote_flags_provider.is_enabled_sync("sample-flag", user_context) + print(f"Sync feature enabled: {is_enabled_sync}") + +if __name__ == '__main__': + asyncio.run(main()) \ No newline at end of file diff --git a/mixpanel/__init__.py b/mixpanel/__init__.py index 046d068..8362dd4 100644 --- a/mixpanel/__init__.py +++ b/mixpanel/__init__.py @@ -24,11 +24,13 @@ from requests.auth import HTTPBasicAuth import urllib3 -__version__ = '4.11.0' -VERSION = __version__ # TODO: remove when bumping major version. +from .flags.local_feature_flags import LocalFeatureFlagsProvider +from .flags.remote_feature_flags import RemoteFeatureFlagsProvider +from .flags.types import LocalFlagsConfig, RemoteFlagsConfig -logger = logging.getLogger(__name__) +__version__ = '5.0.0-rc1' +logger = logging.getLogger(__name__) class DatetimeSerializer(json.JSONEncoder): def default(self, obj): @@ -70,6 +72,23 @@ def _now(self): def _make_insert_id(self): return uuid.uuid4().hex + def getLocalFlagsProvider(self, config: LocalFlagsConfig) -> LocalFeatureFlagsProvider: + """Create and return a local feature flags provider. + + :param LocalFlagsConfig config: Configuration for the local flags provider + :return: LocalFeatureFlagsProvider instance + """ + return LocalFeatureFlagsProvider(self._token, config, self.track) + + def getRemoteFlagsProvider(self, config: RemoteFlagsConfig) -> RemoteFeatureFlagsProvider: + """Create and return a remote feature flags provider. + + :param RemoteFlagsConfig config: Configuration for the remote flags provider + :return: RemoteFeatureFlagsProvider instance + """ + return RemoteFeatureFlagsProvider(self._token, config, self.track) + + def track(self, distinct_id, event_name, properties=None, meta=None): """Record an event. diff --git a/mixpanel/flags/__init__.py b/mixpanel/flags/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mixpanel/flags/local_feature_flags.py b/mixpanel/flags/local_feature_flags.py new file mode 100644 index 0000000..3c2dc6b --- /dev/null +++ b/mixpanel/flags/local_feature_flags.py @@ -0,0 +1,203 @@ +import httpx +import logging +import asyncio +import time +from datetime import datetime, timedelta +from typing import Dict, Any, Callable, Optional +from .types import ExperimentationFlag, ExperimentationFlags, SelectedVariant, LocalFlagsConfig, Rollout +from .utils import REQUEST_HEADERS, normalized_hash, track_exposure_event + +logger = logging.getLogger(__name__) +logging.getLogger("httpx").setLevel(logging.ERROR) + +class LocalFeatureFlagsProvider: + FLAGS_DEFINITIONS_URL_PATH = "/flags/definitions" + + def __init__(self, token: str, config: LocalFlagsConfig, tracker: Callable) -> None: + self._token: str = token + self._config: LocalFlagsConfig = config + self._tracker: Callable = tracker + + self._flag_definitions: Dict[str, ExperimentationFlag] = dict() + + httpx_client_parameters = { + "base_url": f"https://{config.api_host}", + "headers": REQUEST_HEADERS, + "auth": httpx.BasicAuth(token, ""), + "timeout": httpx.Timeout(config.requestTimeoutInSeconds), + } + + self.async_client: httpx.AsyncClient = httpx.AsyncClient(**httpx_client_parameters) + + async def start_polling_for_definitions(self): + self._polling_task = asyncio.create_task(self._poll_for_definitions()) + + async def _poll_for_definitions(self): + await self._fetch_flag_definitions() + + if self._config.enablePolling: + logging.info(f"Initialized async polling for flag definition updates every {self._config.pollingIntervalInSeconds} seconds") + while True: + await asyncio.sleep(self._config.pollingIntervalInSeconds) + await self._fetch_flag_definitions() + + 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. + """ + return bool(self._flag_definitions) + + def get_variant_value(self, flag_key: str, fallback_value: str, context: Dict[str, str]) -> str: + variant = self.get_variant(flag_key, SelectedVariant(variant_key=fallback_value, variant_value=fallback_value), context) + return variant.variant_value + + def is_enabled(self, flag_key: str, context: Dict[str, str]) -> bool: + variant = self.get_variant_value(flag_key, False, context) + return bool(variant) + + def get_variant(self, flag_key: str, fallback_value: SelectedVariant, context: Dict[str, str]) -> SelectedVariant: + start_time = time.perf_counter() + flag_definition = self._flag_definitions.get(flag_key) + + if not flag_definition: + logger.warning(f"Cannot find flag definition for key: {flag_key}") + return fallback_value + + if not(context_value := context.get(flag_definition.context)): + logger.warning(f"The rollout context, {flag_definition.context} for flag, {flag_key} is not present in the supplied context dictionary") + return fallback_value + + if test_user_variant := self._get_variant_override_for_test_user(flag_definition, context): + return test_user_variant + + 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) + return variant + + logger.info(f"{flag_definition.context} context {context_value} not eligible for any rollout for flag: {flag_key}") + return fallback_value + + def _get_variant_override_for_test_user(self, flag_definition: ExperimentationFlag, context: Dict[str, str]) -> Optional[SelectedVariant]: + """""" + if not flag_definition.ruleset.test or not flag_definition.ruleset.test.users: + return None + + if not (distinct_id := context.get("distinct_id")): + return None + + if not (variant_key := flag_definition.ruleset.test.users.get(distinct_id)): + return None + + for variant in flag_definition.ruleset.variants: + if variant_key.casefold() == variant.key.casefold(): + return SelectedVariant(variant_key=variant.key, variant_value=variant.value) + + return None + + def _get_assigned_variant(self, flag_definition: ExperimentationFlag, context_value: Any, flagName: str, rollout: Rollout) -> SelectedVariant: + if rollout.variant_override: + variant = rollout.variant_override + return SelectedVariant(variant_key=variant.key, variant_value=variant.value) + + variants = flag_definition.ruleset.variants + hash_input = str(context_value) + flagName + + variant_hash = normalized_hash(hash_input, "variant") + + selected = None + last = None + cumulative = 0.0 + for variant in variants: + last = variant + cumulative += variant.split + if variant_hash < cumulative: + selected = variant + break + + chosen_variant = selected if selected else last + + return SelectedVariant(variant_key=chosen_variant.key, variant_value=chosen_variant.value) + + def _get_assigned_rollout(self, flag_definition: ExperimentationFlag, context_value: Any, context: Dict[str, str]) -> Optional[Rollout]: + hash_input = str(context_value) + flag_definition.key + + rollout_hash = normalized_hash(hash_input, "rollout") + + for rollout in flag_definition.ruleset.rollout: + if rollout_hash < rollout.rollout_percentage and self._is_runtime_evaluation_satisfied(rollout, context): + return rollout + + return None + + def _is_runtime_evaluation_satisfied(self, rollout: Rollout, context: Dict[str, str]) -> bool: + if not rollout.runtime_evaluation_definition: + return True + + if not (custom_properties := context.get("custom_properties")): + return False + + if not isinstance(custom_properties, dict): + return False + + for key, expected_value in rollout.runtime_evaluation_definition.items(): + if key not in custom_properties: + return False + + actual_value = custom_properties[key] + if actual_value.casefold() != expected_value.casefold(): + return False + + return True + + async def _fetch_flag_definitions(self) -> None: + try: + start_time = datetime.now() + response = await self.async_client.get(self.FLAGS_DEFINITIONS_URL_PATH) + end_time = datetime.now() + request_duration: timedelta = end_time - start_time + logging.info(f"Request started at {start_time.isoformat()}, completed at {end_time.isoformat()}, duration: {request_duration.total_seconds():.3f}s") + + response.raise_for_status() + + flags = {} + try: + json_data = response.json() + experimentation_flags = ExperimentationFlags.model_validate(json_data) + for flag in experimentation_flags.flags: + flags[flag.key] = flag + except Exception as e: + logger.error("Failed to parse flag definitions: {}".format(e)) + + self._flag_definitions = flags + logger.info("Successfully fetched {} flag definitions".format(len(self._flag_definitions))) + except Exception as e: + logger.error("Failed to fetch feature flag definitions: {}".format(e)) + + def track_exposure(self, flag_key: str, variant: SelectedVariant, latencyInSeconds: float, context: Dict[str, str]): + additional_properties = { + "Flag evaluation mode": "remote", + "Variant fetch latency (ms)": latencyInSeconds * 1000 + } + + if distinct_id := context.get("distinct_id"): + track_exposure_event( + distinct_id=distinct_id, + flag_key=flag_key, + variant=variant, + additional_properties=additional_properties, + tracker=self._tracker) + else: + logging.error("Cannot track exposure event without a distinct_id in the context") + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + logging.info("Exiting the LocalFeatureFlagsProvider and cleaning up resources") + if self._polling_task and not self._polling_task.done(): + await self._polling_task.cancel() + + await self.async_client.aclose() diff --git a/mixpanel/flags/remote_feature_flags.py b/mixpanel/flags/remote_feature_flags.py new file mode 100644 index 0000000..414348d --- /dev/null +++ b/mixpanel/flags/remote_feature_flags.py @@ -0,0 +1,120 @@ +import httpx +import logging +import json +import urllib.parse +from datetime import datetime, timedelta +from typing import Dict, Any, Callable + +from .types import RemoteFlagsConfig, SelectedVariant, RemoteFlagsResponse +from .utils import REQUEST_HEADERS, track_exposure_event + +class RemoteFeatureFlagsProvider: + FLAGS_URL_PATH = "/flags" + + def __init__(self, token: str, config: RemoteFlagsConfig, tracker: Callable) -> None: + self._config: RemoteFlagsConfig = config + self._tracker: Callable = tracker + + httpx_client_parameters = { + "base_url": f"https://{config.api_host}", + "headers": REQUEST_HEADERS, + "auth": httpx.BasicAuth(token, ""), + "timeout": httpx.Timeout(config.requestTimeoutInSeconds), + } + + self.async_client: httpx.AsyncClient = httpx.AsyncClient(**httpx_client_parameters) + self.sync_client: httpx.Client = httpx.Client(**httpx_client_parameters) + + async def get_variant_value(self, flag_key: str, fallback_value: str, context: Dict[str, str]) -> SelectedVariant: + variant = await self.get_variant(flag_key, SelectedVariant(variant_key=fallback_value, variant_value=fallback_value), context) + return variant.variant_value + + async def get_variant(self, flag_key: str, fallback_value: SelectedVariant, context: Dict[str, str]) -> SelectedVariant: + try: + params = self._prepare_query_params(flag_key, context) + start_time = datetime.now() + response = await self.async_client.get(self.FLAGS_URL_PATH, params=params) + end_time = datetime.now() + return self._handle_response(context, flag_key, fallback_value, response, start_time, end_time) + except Exception: + logging.exception(f"Failed to get remote variant for flag {flag_key}") + return fallback_value + + async def is_enabled(self, flag_key: str, context: Dict[str, str]) -> bool: + variant = await self.get_variant_value(flag_key, "false", context) + return bool(variant.variant_value) + + def get_variant_value_sync(self, flag_key: str, fallback_value: Any, context: Dict[str, str]) -> SelectedVariant: + variant = self.get_variant_sync(flag_key, SelectedVariant(variant_key=fallback_value, variant_value=fallback_value), context) + return variant.variant_value + + def get_variant_sync(self, flag_key: str, fallback_value: SelectedVariant, context: Dict[str, str]) -> SelectedVariant: + try: + params = self._prepare_query_params(flag_key, context) + start_time = datetime.now() + response = self.sync_client.get(self._url, params=params) + end_time = datetime.now() + return self._handle_response(context, flag_key, fallback_value, response, start_time, end_time) + except Exception: + logging.exception(f"Failed to get remote variant for flag {flag_key}") + return fallback_value + + def is_enabled_sync(self, flag_key: str, context: Dict[str, str]) -> bool: + variant = self.get_variant_value_sync(flag_key, "false", context) + return bool(variant.variant_value) + + def _prepare_query_params(self, flag_key: str, context: Dict[str, str]) -> Dict[str, str]: + context_json = json.dumps(context).encode('utf-8') + url_encoded_context = urllib.parse.quote(context_json) + params = { + 'flag_key': flag_key, + "context": url_encoded_context + } + return params + + def _handle_response(self, context: Dict[str, str], flag_key: str, fallback_value: SelectedVariant, response: httpx.Response, start_time: datetime, end_time: datetime) -> SelectedVariant: + print("The URL we are calling is", self.async_client.base_url) + print("The response content is", response.content) + request_duration: timedelta = (end_time - start_time) + formatted_start_time, formatted_end_time = start_time.isoformat(), end_time.isoformat() + logging.info(f"Request started at {formatted_start_time}, completed at {formatted_end_time}, duration: {request_duration.total_seconds():.3f}s") + + response.raise_for_status() + + flags_response = RemoteFlagsResponse.model_validate(response.json()) + + if flag_key in flags_response.flags: + selected_variant = flags_response.flags[flag_key] + + additional_properties = { + "Flag evaluation mode": "remote", + "Variant fetch start time": formatted_start_time, + "Variant fetch complete time": formatted_end_time, + "Variant fetch latency (ms)": request_duration.total_seconds() * 1000, + } + + if distinct_id := context.get("distinct_id"): + track_exposure_event( + distinct_id=distinct_id, + flag_key=flag_key, + variant=selected_variant, + additional_properties=additional_properties, + tracker=self._tracker) + + return selected_variant + else: + logging.warning(f"Flag {flag_key} not found in remote response. Returning fallback, {fallback_value}") + return fallback_value + + def __enter__(self): + return self + + async def __aenter__(self): + return self + + def __exit__(self): + self.sync_client.close() + + async def __aexit__(self, exc_type, exc_val, exc_tb): + logging.info("Exiting the RemoteFeatureFlagsProvider and cleaning up resources") + await self.async_client.aclose() diff --git a/mixpanel/flags/test_local_feature_flags.py b/mixpanel/flags/test_local_feature_flags.py new file mode 100644 index 0000000..0c7afeb --- /dev/null +++ b/mixpanel/flags/test_local_feature_flags.py @@ -0,0 +1,221 @@ +import pytest +import asyncio +import respx +import httpx +from unittest.mock import Mock, patch +from typing import Dict +from .types import LocalFlagsConfig, ExperimentationFlag, RuleSet, Variant, Rollout, TestUsers, SelectedVariant, ExperimentationFlags +from .local_feature_flags import LocalFeatureFlagsProvider + +@pytest.mark.asyncio +class TestLocalFeatureFlagsProvider: + async def setup_flags(self, flags): + """Helper to setup the provider with specific flags""" + respx.get("https://api.mixpanel.com/flags/definitions").mock( + return_value=self.create_flags_response(flags)) + config = LocalFlagsConfig() + config.enablePolling = False + mock_tracker = Mock() + flags_provider = LocalFeatureFlagsProvider("test-token", config, mock_tracker) + await flags_provider.start_polling_for_definitions() + return flags_provider + + @staticmethod + def create_test_flag( + flag_key: str = "test_flag", + context: str = "distinct_id", + variants: list = None, + rollout_percentage: float = 100.0, + runtime_evaluation: Dict = None, + test_users: Dict[str, str] = None) -> ExperimentationFlag: + + if variants is None: + variants = [ + Variant(key="control", value="control", is_control=True, split=50.0), + Variant(key="treatment", value="treatment", is_control=False, split=50.0) + ] + + rollouts = [Rollout( + rollout_percentage=rollout_percentage, + runtime_evaluation_definition=runtime_evaluation, + variant_override=None + )] + + test_config = None + if test_users: + test_config = TestUsers(users=test_users) + + ruleset = RuleSet( + variants=variants, + rollout=rollouts, + test=test_config + ) + + return ExperimentationFlag( + id="test-id", + name="Test Flag", + key=flag_key, + status="active", + project_id=123, + ruleset=ruleset, + context=context + ) + + def create_flags_response(self, flags: list = None) -> Dict: + """Helper to create API response with flags""" + if flags is None: + flags = [] + response_data = ExperimentationFlags(flags=flags).model_dump() + return httpx.Response(status_code=200, json=response_data) + + @respx.mock + async def test_get_variant_value_returns_fallback_when_no_flag_definitions(self): + flags = await self.setup_flags([]) + result = flags.get_variant_value("nonexistent_flag", "control", {"distinct_id": "user123"}) + assert result == "control" + + @respx.mock + async def test_get_variant_value_returns_fallback_when_flag_does_not_exist(self): + other_flag = self.create_test_flag("other_flag") + flags = await self.setup_flags([other_flag]) + + result = flags.get_variant_value("nonexistent_flag", "control", {"distinct_id": "user123"}) + assert result == "control" + + @respx.mock + async def test_get_variant_value_returns_fallback_when_no_context(self): + flag = self.create_test_flag(context="distinct_id") + flags = await self.setup_flags([flag]) + + result = flags.get_variant_value("test_flag", "fallback", {}) + assert result == "fallback" + + @respx.mock + async def test_get_variant_value_returns_fallback_when_wrong_context_key(self): + flag = self.create_test_flag(context="user_id") + flags = await self.setup_flags([flag]) + + result = flags.get_variant_value("test_flag", "fallback", {"distinct_id": "user123"}) + assert result == "fallback" + + @respx.mock + async def test_get_variant_value_returns_test_user_variant_when_configured(self): + variants = [ + Variant(key="control", value="false", is_control=True, split=50.0), + Variant(key="treatment", value="true", is_control=False, split=50.0) + ] + flag = self.create_test_flag( + variants=variants, + test_users={"test_user": "treatment"} + ) + + flags = await self.setup_flags([flag]) + + result = flags.get_variant_value("test_flag", "control", {"distinct_id": "test_user"}) + + assert result == "true" + + @respx.mock + async def test_get_variant_value_returns_fallback_when_test_user_variant_not_configured(self): + variants = [ + Variant(key="control", value="false", is_control=True, split=50.0), + Variant(key="treatment", value="true", is_control=False, split=50.0) + ] + flag = self.create_test_flag( + variants=variants, + test_users={"test_user": "nonexistent_variant"} + ) + flags = await self.setup_flags([flag]) + + with patch('mixpanel.flags.utils.normalized_hash') as mock_hash: + mock_hash.return_value = 0.5 + result = flags.get_variant_value("test_flag", "fallback", {"distinct_id": "test_user"}) + assert result == "false" + + @respx.mock + async def test_get_variant_value_returns_fallback_when_rollout_percentage_zero(self): + flag = self.create_test_flag(rollout_percentage=0.0) + flags = await self.setup_flags([flag]) + + result = flags.get_variant_value("test_flag", "fallback", {"distinct_id": "user123"}) + assert result == "fallback" + + @respx.mock + async def test_get_variant_value_returns_variant_when_rollout_percentage_hundred(self): + flag = self.create_test_flag(rollout_percentage=100.0) + flags = await self.setup_flags([flag]) + + with patch('mixpanel.flags.utils.normalized_hash') as mock_hash: + mock_hash.return_value = 0.5 + result = flags.get_variant_value("test_flag", "fallback", {"distinct_id": "user123"}) + assert result != "fallback" + assert result in ["false", "true"] + + @respx.mock + async def test_get_variant_value_respects_runtime_evaluation_satisfied(self): + runtime_eval = {"plan": "premium", "region": "US"} + flag = self.create_test_flag(runtime_evaluation=runtime_eval) + flags = await self.setup_flags([flag]) + + context = { + "distinct_id": "user123", + "custom_properties": { + "plan": "premium", + "region": "US" + } + } + + with patch('mixpanel.flags.utils.normalized_hash') as mock_hash: + mock_hash.return_value = 0.5 + result = flags.get_variant_value("test_flag", "fallback", context) + assert result != "fallback" + + @respx.mock + async def test_get_variant_value_returns_fallback_when_runtime_evaluation_not_satisfied(self): + runtime_eval = {"plan": "premium", "region": "US"} + flag = self.create_test_flag(runtime_evaluation=runtime_eval) + flags = await self.setup_flags([flag]) + + context = { + "distinct_id": "user123", + "custom_properties": { + "plan": "basic", + "region": "US" + } + } + + result = flags.get_variant_value("test_flag", "fallback", context) + assert result == "fallback" + + @respx.mock + async def test_get_variant_value_picks_correct_variant_with_hundred_percent_split(self): + variants = [ + Variant(key="A", value="variant_a", is_control=False, split=100.0), + Variant(key="B", value="variant_b", is_control=False, split=0.0), + Variant(key="C", value="variant_c", is_control=False, split=0.0) + ] + flag = self.create_test_flag(variants=variants, rollout_percentage=100.0) + flags = await self.setup_flags([flag]) + + with patch('mixpanel.flags.utils.normalized_hash') as mock_hash: + for hash_val in [0.1, 0.5, 0.9]: + mock_hash.return_value = hash_val + result = flags.get_variant_value("test_flag", "fallback", {"distinct_id": "user123"}) + assert result == "variant_a" + + @respx.mock + async def test_get_variant_value_tracks_exposure_when_variant_selected(self): + flag = self.create_test_flag() + flags = await self.setup_flags([flag]) + + 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._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._tracker.assert_not_called() \ No newline at end of file diff --git a/mixpanel/flags/test_remote_feature_flags.py b/mixpanel/flags/test_remote_feature_flags.py new file mode 100644 index 0000000..1c82f90 --- /dev/null +++ b/mixpanel/flags/test_remote_feature_flags.py @@ -0,0 +1,72 @@ +import pytest +import httpx +import respx +from typing import Dict +from unittest.mock import Mock +from .types import RemoteFlagsConfig, ExperimentationFlags, RemoteFlagsResponse, SelectedVariant +from .remote_feature_flags import RemoteFeatureFlagsProvider + +ENDPOINT = "https://api.mixpanel.com/flags" + +@pytest.mark.asyncio +class TestRemoteFeatureFlagsProvider: + def setup_method(self): + config = RemoteFlagsConfig() + mock_tracker = Mock() + self._flags = RemoteFeatureFlagsProvider("test-token", config, mock_tracker) + + @staticmethod + def create_success_response(assignedVariantsPerFlag: Dict[str, SelectedVariant]) -> Dict: + serialized_response = RemoteFlagsResponse(code=200, flags=assignedVariantsPerFlag).model_dump() + return httpx.Response(status_code=200, json=serialized_response) + + @respx.mock + async def test_get_variant_value_is_fallback_if_call_fails(self): + respx.get(ENDPOINT).mock(side_effect=httpx.RequestError("Network error")) + + result = await self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"}) + + assert result == "control" + + @respx.mock + async def test_get_variant_value_is_fallback_if_bad_response_format(self): + respx.get(ENDPOINT).mock(return_value=httpx.Response(200, text="invalid json")) + + result = await self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"}) + + assert result == "control" + + @respx.mock + async def test_get_variant_value_is_fallback_if_success_but_no_flag_found(self): + respx.get(ENDPOINT).mock( + return_value=self.create_success_response({})) + + result = await self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"}) + + assert result == "control" + + @respx.mock + async def test_get_variant_value_returns_expected_variant_from_api(self): + respx.get(ENDPOINT).mock( + return_value=self.create_success_response({"test_flag": SelectedVariant(variant_key="treatment", variant_value="treatment")})) + + result = await self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"}) + + assert result == "treatment" + + @respx.mock + async def test_get_variant_value_tracks_exposure_event_if_variant_selected(self): + respx.get(ENDPOINT).mock( + return_value=self.create_success_response({"test_flag": SelectedVariant(variant_key="treatment", variant_value="treatment")})) + + await self._flags.get_variant_value("test_flag", "control", {"distinct_id": "user123"}) + + self._flags._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.get_variant_value("test_flag", "control", {"distinct_id": "user123"}) + + self._flags._tracker.assert_not_called() \ No newline at end of file diff --git a/mixpanel/flags/types.py b/mixpanel/flags/types.py new file mode 100644 index 0000000..be77088 --- /dev/null +++ b/mixpanel/flags/types.py @@ -0,0 +1,55 @@ +from typing import Optional, List, Dict, Any +from pydantic import BaseModel + +MIXPANEL_DEFAULT_API_ENDPOINT = "api.mixpanel.com" +class FlagsConfig(BaseModel): + api_host: str = "api.mixpanel.com" + requestTimeoutInSeconds: int = 10 + retryLimit: int = 3 + retryExponentialBackoffFactor: int = 1 + +class LocalFlagsConfig(FlagsConfig): + enablePolling: bool = True + pollingIntervalInSeconds: int = 60 + +class RemoteFlagsConfig(FlagsConfig): + pass + +class Variant(BaseModel): + key: str + value: str + is_control: bool + split: float + +class TestUsers(BaseModel): + users: Dict[str, str] + +class Rollout(BaseModel): + rollout_percentage: float + runtime_evaluation_definition: Optional[Dict[str, Any]] + variant_override: Optional[Variant] + +class RuleSet(BaseModel): + variants: List[Variant] + rollout: List[Rollout] + test: Optional[TestUsers] + +class ExperimentationFlag(BaseModel): + id: str + name: str + key: str + status: str + project_id: int + ruleset: RuleSet + context: str + +class SelectedVariant(BaseModel): + variant_key: str + variant_value: str + +class ExperimentationFlags(BaseModel): + flags: List[ExperimentationFlag] + +class RemoteFlagsResponse(BaseModel): + code: int + flags: Dict[str, SelectedVariant] \ No newline at end of file diff --git a/mixpanel/flags/utils.py b/mixpanel/flags/utils.py new file mode 100644 index 0000000..f1dc9e4 --- /dev/null +++ b/mixpanel/flags/utils.py @@ -0,0 +1,59 @@ +from typing import Dict +from .types import SelectedVariant + +REQUEST_HEADERS: Dict[str, str] = { + 'X-Scheme': 'https', + 'X-Forwarded-Proto': 'https', + 'Content-Type': 'application/json' +} + +def track_exposure_event( + distinct_id: str, + flag_key: str, + variant: SelectedVariant, + additional_properties: Dict[str, str], + tracker) -> None: + """Tracks an experiment started event to Mixpanel + + :param distinct_id: The distinct ID of the user being tracked + :param flag: The experimentation flag being tracked. + :param variant: The selected variant for the experiment + :param additional_properties: Additional properties to include in the event + :param tracker: The tracker function to use for sending the event + """ + + properties = { + 'Experiment name': flag_key, + 'Variant name': variant.variant_key, + '$experiment_type': 'feature_flag', + } + + properties.update(additional_properties) + + tracker(distinct_id, '$experiment_started', properties) + +def normalized_hash(key: str, salt: str) -> float: + """Compute a normalized hash using FNV-1a algorithm. + + :param key: The key to hash + :param salt: Salt to add to the hash + :return: Normalized hash value between 0.0 and 1.0 + """ + hash_value = _fnv1a64(key.encode("utf-8") + salt.encode("utf-8")) + return (hash_value % 100) / 100.0 + +def _fnv1a64(data: bytes) -> int: + """FNV-1a 64-bit hash function. + + :param data: Bytes to hash + :return: 64-bit hash value + """ + FNV_prime = 0x100000001b3 + hash_value = 0xcbf29ce484222325 + + for byte in data: + hash_value ^= byte + hash_value *= FNV_prime + hash_value &= 0xffffffffffffffff # Keep it 64-bit + + return hash_value \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 966a674..b614eb6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,8 @@ authors = [ requires-python = ">=3.9" dependencies = [ "requests>=2.32.5", + "httpx>=0.27.0", + "pydantic>=2.0.0", ] keywords = ["mixpanel", "analytics"] classifiers = [ @@ -32,7 +34,9 @@ Homepage = "https://github.com/mixpanel/mixpanel-python" [project.optional-dependencies] test = [ "pytest>=8.4.1", + "pytest-asyncio>=0.23.0", "responses>=0.25.8", + "respx>=0.21.0", ] dev = [ "tox>=4.28.4", @@ -55,4 +59,7 @@ envlist = ["py39", "py310", "py311", "py312"] extras = ["test"] commands = [ ["pytest", "{posargs}"], -] \ No newline at end of file +] + +[tool.pytest.ini_options] +asyncio_mode = "auto" \ No newline at end of file