Add disable_ua_parser and disable_country_lookup options to reduce memory usage (#454)

using the kong eval project key - data for how much mem used for
initialize
UA + Country enabled: +24.2 MiB 
<img width="566" height="126" alt="Screenshot 2025-08-14 at 11 29 20 AM"
src="https://github.com/user-attachments/assets/e042a5f8-c267-41e7-a25a-87eabc6c9b5e"
/>
Only UA disabled: +20.5 MiB
<img width="631" height="219" alt="Screenshot 2025-08-14 at 11 57 10 AM"
src="https://github.com/user-attachments/assets/1f2e19d0-05a8-4be4-9d91-8c5136df90cb"
/>
Only Country disabled: +8.6 MiB
<img width="670" height="186" alt="Screenshot 2025-08-14 at 12 07 52 PM"
src="https://github.com/user-attachments/assets/50b4300f-1b2c-42fb-8114-b4a4cca51abb"
/>
Disabling both: +6.7 MiB
<img width="673" height="214" alt="Screenshot 2025-08-14 at 12 08 49 PM"
src="https://github.com/user-attachments/assets/efda6376-258c-4259-bfb7-b5899675fa2a"
/>




Adds two new boolean options to `StatsigOptions` that allow users to
disable UA parsing and country lookup functionality to reduce memory
usage. When disabled, these features will print warning messages if
accessed during evaluation but will not perform the actual
parsing/lookup operations.

**Link to Devin run**:
https://app.devin.ai/sessions/3d3125c33eac43ecb454985bfd56ec59
**Requested by**: [email protected]

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: [email protected] <[email protected]>

Author: devin-ai-integration[bot]

statsig/evaluation_details.py

@@ -16,6 +16,8 @@ class DataSource(str, Enum):
     NETWORK = "Network"
     STATSIG_NETWORK = "StatsigNetwork"
     UNINITIALIZED = "Uninitialized"
+    UA_NOT_LOADED = "UAParserNotLoaded"
+    COUNTRY_NOT_LOADED = "CountryLookupNotLoaded"
 
 
 class EvaluationDetails:

statsig/evaluator.py

@@ -6,7 +6,6 @@
 from typing import Any, Dict, Optional, Union
 
 from ip3country import CountryLookup
-from ua_parser import user_agent_parser
 
 from .client_initialize_formatter import ClientInitializeResponseFormatter
 from .config_evaluation import _ConfigEvaluation
@@ -17,18 +16,34 @@
 from .utils import HashingAlgorithm, JSONValue, sha256_hash
 
 
+def load_ua_parser():
+    try:
+        from ua_parser import user_agent_parser  # pylint: disable=import-outside-toplevel
+        return user_agent_parser
+    except ImportError:
+        logger.warning("ua_parser module not available")
+        return None
+
+
 class _Evaluator:
-    def __init__(self, spec_store: _SpecStore, global_custom_fields: Optional[Dict[str, JSONValue]]):
+    def __init__(self, spec_store: _SpecStore, global_custom_fields: Optional[Dict[str, JSONValue]],
+                 disable_ua_parser: bool = False, disable_country_lookup: bool = False):
         self._spec_store = spec_store
         self._global_custom_fields = global_custom_fields
+        self._disable_ua_parser = disable_ua_parser
+        self._disable_country_lookup = disable_country_lookup
 
         self._country_lookup: Optional[CountryLookup] = None
+        self._ua_parser: Optional[Any] = None  # Will be the ua_parser.user_agent_parser module
         self._gate_overrides: Dict[str, dict] = {}
         self._config_overrides: Dict[str, dict] = {}
         self._layer_overrides: Dict[str, dict] = {}
 
     def initialize(self):
-        self._country_lookup = CountryLookup()
+        if not self._disable_country_lookup:
+            self._country_lookup = CountryLookup()
+        if not self._disable_ua_parser:
+            self._ua_parser = load_ua_parser()
 
     def override_gate(self, gate, value, user_id=None):
         gate_overrides = self._gate_overrides.get(gate)
@@ -120,6 +135,19 @@ def _create_evaluation_details(self,
         return EvaluationDetails(
             self._spec_store.last_update_time(), self._spec_store.initial_update_time, source, reason)
 
+
+    def _update_evaluation_details_if_needed(self, end_result: _ConfigEvaluation) -> None:
+        current_details = end_result.evaluation_details
+
+        if current_details is None:
+            end_result.evaluation_details = self._create_evaluation_details()
+            return
+
+        if current_details.source in [DataSource.UA_NOT_LOADED, DataSource.COUNTRY_NOT_LOADED]:
+            return
+
+        end_result.evaluation_details = self._create_evaluation_details()
+
     def __lookup_gate_override(self, user, gate):
         gate_overrides = self._gate_overrides.get(gate)
         if gate_overrides is None:
@@ -336,7 +364,11 @@ def __finalize_eval_result(self, config, end_result, did_pass, rule, is_nested=F
         if config.get("version", None) is not None:
             end_result.version = config.get("version")
 
-        end_result.evaluation_details = self._create_evaluation_details()
+        if end_result.evaluation_details is not None and end_result.evaluation_details.source not in (
+        DataSource.UA_NOT_LOADED, DataSource.COUNTRY_NOT_LOADED):
+            end_result.evaluation_details = self._create_evaluation_details()
+
+        self._update_evaluation_details_if_needed(end_result)
 
         if rule is None:
             end_result.json_value = config.get("defaultValue", {})
@@ -436,16 +468,22 @@ def __evaluate_condition(self, user, condition, end_result, sampling_rate=None):
             if value is None:
                 ip = self.__get_from_user(user, "ip")
                 if ip is not None and field == "country":
-                    if not self._country_lookup:
-                        self._country_lookup = CountryLookup()
-                    value = self._country_lookup.lookupStr(ip)
+                    if self._disable_country_lookup:
+                        logger.warning("Country lookup is disabled but was attempted during evaluation")
+                        end_result.evaluation_details = self._create_evaluation_details(
+                            EvaluationReason.none, DataSource.COUNTRY_NOT_LOADED)
+                        value = None
+                    else:
+                        if not self._country_lookup:
+                            self._country_lookup = CountryLookup()
+                        value = self._country_lookup.lookupStr(ip)
             if value is None:
                 end_result.analytical_condition = sampling_rate is None
                 return False
         elif type == "UA_BASED":
             value = self.__get_from_user(user, field)
             if value is None:
-                value = self.__get_from_user_agent(user, field)
+                value = self.__get_from_user_agent(user, field, end_result)
         elif type == "USER_FIELD":
             value = self.__get_from_user(user, field)
         elif type == "CURRENT_TIME":
@@ -754,11 +792,26 @@ def __get_value_as_float(self, input):
             return None
         return float(input)
 
-    def __get_from_user_agent(self, user, field):
+    def __get_from_user_agent(self, user, field, end_result):
+        if self._disable_ua_parser:
+            logger.warning("UA parser is disabled but was attempted during evaluation")
+            end_result.evaluation_details = self._create_evaluation_details(EvaluationReason.none,
+                                                                            DataSource.UA_NOT_LOADED)
+            return None
         ua = self.__get_from_user(user, "userAgent")
         if ua is None:
             return None
-        parsed = user_agent_parser.Parse(ua)
+
+        try:
+            if self._ua_parser is None:
+                self._ua_parser = load_ua_parser()
+            if self._ua_parser is None:
+                return None
+            parsed = self._ua_parser.Parse(ua)
+        except Exception as e:
+            logger.warning(f"Error parsing user agent: {e}")
+            return None
+
         field = field.lower()
         if field in ("osname", "os_name"):
             return parsed.get("os", {"family": None}).get("family")

statsig/statsig_options.py

@@ -116,6 +116,8 @@ def __init__(
             events_flushed_callback: Optional[
                 Callable[[bool, List[Dict], Optional[int], Optional[Exception]], None]] = None,
             global_custom_fields: Optional[Dict[str, JSONValue]] = None,
+            disable_ua_parser: bool = False,
+            disable_country_lookup: bool = False,
     ):
         self.data_store = data_store
         self._environment: Union[None, dict] = None
@@ -167,6 +169,8 @@ def __init__(
         self.events_flushed_callback = events_flushed_callback
         self._logging_copy: Dict[str, Any] = {}
         self.global_custom_fields = global_custom_fields
+        self.disable_ua_parser = disable_ua_parser
+        self.disable_country_lookup = disable_country_lookup
         self._set_logging_copy()
         self._attributes_changed = False
 
@@ -250,5 +254,9 @@ def _set_logging_copy(self):
             logging_copy["sdk_error_callback"] = "SET"
         if self.global_custom_fields:
             logging_copy["global_custom_fields"] = to_raw_dict_or_none(self.global_custom_fields)
+        if self.disable_ua_parser:
+            logging_copy["disable_ua_parser"] = self.disable_ua_parser
+        if self.disable_country_lookup:
+            logging_copy["disable_country_lookup"] = self.disable_country_lookup
         self._logging_copy = logging_copy
         self._attributes_changed = False

statsig/statsig_server.py

@@ -124,7 +124,7 @@ def _initialize_impl(self, sdk_key: str, options: Optional[StatsigOptions]):
                 init_context
             )
 
-            self._evaluator = _Evaluator(self._spec_store, self._options.global_custom_fields)
+            self._evaluator = _Evaluator(self._spec_store, self._options.global_custom_fields, self._options.disable_ua_parser, self._options.disable_country_lookup)
 
             init_timeout = options.overall_init_timeout
             if init_timeout is not None: