updated client, factory, validator and config classes

Author: chillaq

splitio/client/client.py

@@ -6,7 +6,7 @@
 from splitio.models.impressions import Impression, Label
 from splitio.models.events import Event, EventWrapper
 from splitio.models.telemetry import get_latency_bucket_index, MethodExceptionsAndLatencies
-from splitio.client import input_validator
+from splitio.client import input_validator, config
 from splitio.util.time import get_current_epoch_time_ms, utctime_ms
 
 
@@ -346,6 +346,113 @@ def get_treatments_with_config(self, key, feature_flag_names, attributes=None):
         except Exception:
             return {feature: (CONTROL, None) for feature in feature_flag_names}
 
+    def get_treatments_by_flag_set(self, key, flag_set, attributes=None):
+            """
+            Get treatments for feature flags that contain given flag set.
+            This method never raises an exception. If there's a problem, the appropriate log message
+            will be generated and the method will return the CONTROL treatment.
+            :param key: The key for which to get the treatment
+            :type key: str
+            :param flag_set: flag set
+            :type flag_sets: str
+            :param attributes: An optional dictionary of attributes
+            :type attributes: dict
+            :return: Dictionary with the result of all the feature flags provided
+            :rtype: dict
+            """
+            return self._get_treatments_by_flag_sets( key, [flag_set], MethodExceptionsAndLatencies.TREATMENTS_BY_FLAG_SET, attributes)
+
+    def get_treatments_by_flag_sets(self, key, flag_sets, attributes=None):
+        """
+        Get treatments for feature flags that contain given flag sets.
+        This method never raises an exception. If there's a problem, the appropriate log message
+        will be generated and the method will return the CONTROL treatment.
+        :param key: The key for which to get the treatment
+        :type key: str
+        :param flag_sets: list of flag sets
+        :type flag_sets: list
+        :param attributes: An optional dictionary of attributes
+        :type attributes: dict
+        :return: Dictionary with the result of all the feature flags provided
+        :rtype: dict
+        """
+        return self._get_treatments_by_flag_sets( key, flag_sets, MethodExceptionsAndLatencies.TREATMENTS_BY_FLAG_SETS, attributes)
+
+    def get_treatments_with_config_by_flag_set(self, key, flag_set, attributes=None):
+        """
+        Get treatments for feature flags that contain given flag set.
+        This method never raises an exception. If there's a problem, the appropriate log message
+        will be generated and the method will return the CONTROL treatment.
+        :param key: The key for which to get the treatment
+        :type key: str
+        :param flag_set: flag set
+        :type flag_sets: str
+        :param attributes: An optional dictionary of attributes
+        :type attributes: dict
+        :return: Dictionary with the result of all the feature flags provided
+        :rtype: dict
+        """
+        return self._get_treatments_by_flag_sets( key, [flag_set], MethodExceptionsAndLatencies.TREATMENTS_WITH_CONFIG_BY_FLAG_SET, attributes)
+
+    def get_treatments_with_config_by_flag_sets(self, key, flag_sets, attributes=None):
+        """
+        Get treatments for feature flags that contain given flag set.
+        This method never raises an exception. If there's a problem, the appropriate log message
+        will be generated and the method will return the CONTROL treatment.
+        :param key: The key for which to get the treatment
+        :type key: str
+        :param flag_set: flag set
+        :type flag_sets: str
+        :param attributes: An optional dictionary of attributes
+        :type attributes: dict
+        :return: Dictionary with the result of all the feature flags provided
+        :rtype: dict
+        """
+        return self._get_treatments_by_flag_sets( key, flag_sets, MethodExceptionsAndLatencies.TREATMENTS_WITH_CONFIG_BY_FLAG_SETS, attributes)
+
+    def _get_treatments_by_flag_sets(self, key, flag_sets, method, attributes=None):
+        """
+        Get treatments for feature flags that contain given flag sets.
+        This method never raises an exception. If there's a problem, the appropriate log message
+        will be generated and the method will return the CONTROL treatment.
+        :param key: The key for which to get the treatment
+        :type key: str
+        :param flag_sets: list of flag sets
+        :type flag_sets: list
+        :param method: Treatment by flag set method flavor
+        :type method: splitio.models.telemetry.MethodExceptionsAndLatencies
+        :param attributes: An optional dictionary of attributes
+        :type attributes: dict
+        :return: Dictionary with the result of all the feature flags provided
+        :rtype: dict
+        """
+        feature_flags_names = self._get_feature_flag_names_by_flag_sets(flag_sets, method.value)
+        if feature_flags_names == []:
+            _LOGGER.warning("%s: No valid Flag set or no feature flags found for evaluating treatments" % (method.value))
+            return {}
+
+        if 'config' in method.value:
+            return self._get_treatments(key, feature_flags_names, method, attributes)
+
+        with_config = self._get_treatments(key, feature_flags_names, method, attributes)
+        return {feature_flag: result[0] for (feature_flag, result) in with_config.items()}
+
+
+    def _get_feature_flag_names_by_flag_sets(self, flag_sets, method_name):
+        """
+        Sanitize given flag sets and return list of feature flag names associated with them
+        :param flag_sets: list of flag sets
+        :type flag_sets: list
+        :return: list of feature flag names
+        :rtype: list
+        """
+        sanitized_flag_sets = input_validator.validate_flag_sets(flag_sets, method_name)
+        feature_flags_by_set = self._split_storage.get_feature_flags_by_sets(sanitized_flag_sets)
+        if feature_flags_by_set is None:
+            _LOGGER.warning("Fetching feature flags for flag set %s encountered an error, skipping this flag set." % (flag_sets))
+            return []
+        return feature_flags_by_set
+
     def _get_treatments(self, key, features, method, attributes=None):
         """
         Validate key, feature flag names and objects, and get the treatments and configs with an optional dictionary of attributes.

splitio/client/config.py

@@ -3,7 +3,7 @@
 import logging
 
 from splitio.engine.impressions import ImpressionsMode
-
+from splitio.client.input_validator import validate_flag_sets
 
 _LOGGER = logging.getLogger(__name__)
 DEFAULT_DATA_SAMPLING = 1
@@ -58,7 +58,8 @@
     'dataSampling': DEFAULT_DATA_SAMPLING,
     'storageWrapper': None,
     'storagePrefix': None,
-    'storageType': None
+    'storageType': None,
+    'flagSetsFilter': None
 }
 
 
@@ -143,4 +144,10 @@ def sanitize(sdk_key, config):
         _LOGGER.warning('metricRefreshRate parameter minimum value is 60 seconds, defaulting to 3600 seconds.')
         processed['metricsRefreshRate'] = 3600
 
+    if config['operationMode'] == 'consumer' and config.get('flagSetsFilter') is not None:
+        processed['flagSetsFilter'] = None
+        _LOGGER.warning('config: FlagSets filter is not applicable for Consumer modes where the SDK does keep rollout data in sync. FlagSet filter was discarded.')
+    else:
+        processed['flagSetsFilter'] = sorted(validate_flag_sets(processed['flagSetsFilter'], 'SDK Config')) if processed['flagSetsFilter'] is not None else None
+
     return processed

splitio/client/factory.py

@@ -498,7 +498,8 @@ def _wrap_impression_listener_async(listener, metadata):
     return None
 
 def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-locals
-                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None):
+                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None,
+                             total_flag_sets=0, invalid_flag_sets=0):
     """Build and return a split factory tailored to the supplied config."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
@@ -536,7 +537,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
     }
 
     storages = {
-        'splits': InMemorySplitStorage(),
+        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),
         'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize'], telemetry_runtime_producer),
         'events': InMemoryEventStorage(cfg['eventsQueueSize'], telemetry_runtime_producer),
@@ -607,7 +608,7 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         unique_keys_tracker=unique_keys_tracker
     )
 
-    telemetry_init_producer.record_config(cfg, extra_cfg)
+    telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets)
 
     if preforked_initialization:
         synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
@@ -625,7 +626,8 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
                         telemetry_submitter)
 
 async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=None,  # pylint:disable=too-many-arguments,too-many-localsa
-                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None):
+                             auth_api_base_url=None, streaming_api_base_url=None, telemetry_api_base_url=None,
+                             total_flag_sets=0, invalid_flag_sets=0):
     """Build and return a split factory tailored to the supplied config in async mode."""
     if not input_validator.validate_factory_instantiation(api_key):
         return None
@@ -663,7 +665,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
     }
 
     storages = {
-        'splits': InMemorySplitStorageAsync(),
+        'splits': InMemorySplitStorageAsync(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorageAsync(),
         'impressions': InMemoryImpressionStorageAsync(cfg['impressionsQueueSize'], telemetry_runtime_producer),
         'events': InMemoryEventStorageAsync(cfg['eventsQueueSize'], telemetry_runtime_producer),
@@ -733,7 +735,7 @@ async def _build_in_memory_factory_async(api_key, cfg, sdk_url=None, events_url=
         unique_keys_tracker=unique_keys_tracker
     )
 
-    await telemetry_init_producer.record_config(cfg, extra_cfg)
+    await telemetry_init_producer.record_config(cfg, extra_cfg, total_flag_sets, invalid_flag_sets)
 
     if preforked_initialization:
         await synchronizer.sync_all(max_retry_attempts=_MAX_RETRY_SYNC_ALL)
@@ -814,7 +816,7 @@ def _build_redis_factory(api_key, cfg):
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
     initialization_thread.start()
 
-    telemetry_init_producer.record_config(cfg, {})
+    telemetry_init_producer.record_config(cfg, {}, 0, 0)
 
     split_factory = SplitFactory(
         api_key,
@@ -894,7 +896,7 @@ async def _build_redis_factory_async(api_key, cfg):
     )
 
     manager = RedisManagerAsync(synchronizer)
-    await telemetry_init_producer.record_config(cfg, {})
+    await telemetry_init_producer.record_config(cfg, {}, 0, 0)
     manager.start()
 
     split_factory = SplitFactoryAsync(
@@ -977,7 +979,7 @@ def _build_pluggable_factory(api_key, cfg):
     initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
     initialization_thread.start()
 
-    telemetry_init_producer.record_config(cfg, {})
+    telemetry_init_producer.record_config(cfg, {}, 0, 0)
 
     split_factory = SplitFactory(
         api_key,
@@ -1056,7 +1058,7 @@ async def _build_pluggable_factory_async(api_key, cfg):
     # Using same class as redis for consumer mode only
     manager = RedisManagerAsync(synchronizer)
     manager.start()
-    await telemetry_init_producer.record_config(cfg, {})
+    await telemetry_init_producer.record_config(cfg, {}, 0, 0)
 
     split_factory = SplitFactoryAsync(
         api_key,
@@ -1083,7 +1085,7 @@ def _build_localhost_factory(cfg):
     telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
 
     storages = {
-        'splits': InMemorySplitStorage(),
+        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),  # not used, just to avoid possible future errors.
         'impressions': LocalhostImpressionsStorage(),
         'events': LocalhostEventsStorage(),
@@ -1282,8 +1284,14 @@ async def get_factory_async(api_key, **kwargs):
     _INSTANTIATED_FACTORIES.update([api_key])
     _INSTANTIATED_FACTORIES_LOCK.release()
 
-    config = sanitize_config(api_key, kwargs.get('config', {}))
+    config_raw = kwargs.get('config', {})
+    total_flag_sets = 0
+    invalid_flag_sets = 0
+    if config_raw.get('flagSetsFilter') is not None and isinstance(config_raw.get('flagSetsFilter'), list):
+        total_flag_sets = len(config_raw.get('flagSetsFilter'))
+        invalid_flag_sets = total_flag_sets - len(input_validator.validate_flag_sets(config_raw.get('flagSetsFilter'), 'Telemetry Init'))
 
+    config = sanitize_config(api_key, config_raw)
     if config['operationMode'] == 'localhost':
         split_factory =  await _build_localhost_factory_async(config)
     elif config['storageType'] == 'redis':
@@ -1298,8 +1306,9 @@ async def get_factory_async(api_key, **kwargs):
         kwargs.get('events_api_base_url'),
         kwargs.get('auth_api_base_url'),
         kwargs.get('streaming_api_base_url'),
-        kwargs.get('telemetry_api_base_url'))
-
+        kwargs.get('telemetry_api_base_url'),
+        total_flag_sets,
+        invalid_flag_sets)
     return split_factory
 
 def _get_active_and_redundant_count():