Merge pull request #98 from splitio/feature/ImpressionListener

[SDKS-153]: Add Impression Listener for Python

Author: mmelograno

.gitignore

@@ -23,6 +23,7 @@ var/
 *.egg-info/
 .installed.cfg
 *.egg
+venv/
 .vscode
 
 # PyInstaller

Detailed-README.md

@@ -110,6 +110,42 @@ except TimeoutException:
     sys.exit()
 ```
 
+## Impression Listener
+Split SDKs send impression data back to Split servers periodically and as a result of evaluating splits. In order to additionally send this information to a location of your choice, you could define and attach an Impression Listener. For that purpose, SDK's options have a parameter called `impressionListener` where an implementation of `ImpressionListener` could be added. This implementation **must** define the `log_impression` method and it will receive data in the following schema:
+
+| Name | Type | Description |
+| --- | --- | --- |
+| impression | Impression | Impression object that has the feature_name, treatment result, label, etc. |
+| attributes | Array | A list of attributes passed by the client. |
+| instance-id | String | Corresponds to the IP of the machine where the SDK is running. |
+| sdk-language-version | String | Indicates the version of the sdk. In this case the language will be python plus the version of it. |
+
+### Implementing custom Impression Listener
+Below you could find an example of how implement a custom Impression Listener:
+```python
+# Import ImpressionListener interface
+from splitio.impressions import ImpressionListener
+
+# Implementation Sample for a Custom Impression Listener
+class CustomImpressionListener(ImpressionListener)
+{
+  def log_impression(self, data):
+    # Custom behavior
+}
+```
+
+### Attaching custom Impression Listener
+```python
+factory = get_factory(
+  'YOUR_API_KEY',
+  config={
+    # ...
+    'impressionListener': CustomImpressionListener()
+  },
+  # ...
+)
+split = factory.client()
+
 ## Additional information
 
 You can get more information on how to use this package in the included documentation.

splitio/brokers.py

@@ -57,12 +57,13 @@ class BaseBroker(object):
 
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self):
+    def __init__(self, config=None):
         """
         Class constructor, only sets up the logger
         """
         self._logger = logging.getLogger(self.__class__.__name__)
         self._destroyed = False
+        self._config = config
 
     def fetch_feature(self, name):
         """
@@ -120,7 +121,7 @@ def destroy(self):
         pass
 
 class JSONFileBroker(BaseBroker):
-    def __init__(self, segment_changes_file_name, split_changes_file_name):
+    def __init__(self, config, segment_changes_file_name, split_changes_file_name):
         """
         A Broker implementation that uses responses from the segmentChanges and
         splitChanges resources to provide access to splits. It is intended to be
@@ -133,7 +134,7 @@ def __init__(self, segment_changes_file_name, split_changes_file_name):
             splitChanges response
         :type split_changes_file_name: str
         """
-        super(JSONFileBroker, self).__init__()
+        super(JSONFileBroker, self).__init__(config)
         self._segment_changes_file_name = segment_changes_file_name
         self._split_changes_file_name = split_changes_file_name
         self._split_fetcher = self._build_split_fetcher()
@@ -310,7 +311,6 @@ def _build_treatment_log(self):
             self._sdk_api,
             max_count=self._max_impressions_log_size,
             interval=self._impressions_interval,
-            listener=self._impression_listener
         )
         return AsyncTreatmentLog(self_updating_treatment_log)
 
@@ -388,7 +388,7 @@ class LocalhostEventStorage(object):
         def log(self, event):
             pass
 
-    def __init__(self, split_definition_file_name=None, auto_refresh_period=2):
+    def __init__(self, config, split_definition_file_name=None, auto_refresh_period=2):
         """
         A broker implementation that builds its configuration from a split
         definition file. By default the definition is taken from $HOME/.split
@@ -398,7 +398,7 @@ def __init__(self, split_definition_file_name=None, auto_refresh_period=2):
         :param auto_refresh_period: Number of seconds between split refresh calls
         :type auto_refresh_period: int
         """
-        super(LocalhostBroker, self).__init__()
+        super(LocalhostBroker, self).__init__(config)
 
         if split_definition_file_name is None:
             self._split_definition_file_name = os.path.join(
@@ -503,11 +503,11 @@ def destroy(self):
 
 
 class RedisBroker(BaseBroker):
-    def __init__(self, redis):
+    def __init__(self, redis, config):
         """A Broker implementation that uses Redis as its backend.
         :param redis: A redis broker
         :type redis: StrctRedis"""
-        super(RedisBroker, self).__init__()
+        super(RedisBroker, self).__init__(config)
 
         split_cache = RedisSplitCache(redis)
         split_fetcher = CacheBasedSplitFetcher(split_cache)
@@ -571,7 +571,7 @@ def __init__(self, uwsgi, config=None):
         :param config: The configuration dictionary
         :type config: dict
         """
-        super(UWSGIBroker, self).__init__()
+        super(UWSGIBroker, self).__init__(config)
 
         split_cache = UWSGISplitCache(uwsgi)
         split_fetcher = CacheBasedSplitFetcher(split_cache)
@@ -712,7 +712,7 @@ def get_self_refreshing_broker(api_key, **kwargs):
     )
 
     if api_key == 'localhost':
-        return LocalhostBroker(**kwargs)
+        return LocalhostBroker(config, **kwargs)
 
     return SelfRefreshingBroker(
         api_key,
@@ -776,11 +776,11 @@ def get_redis_broker(api_key, **kwargs):
     api_key, config, _, _ = _init_config(api_key, **kwargs)
 
     if api_key == 'localhost':
-        return LocalhostBroker(**kwargs)
+        return LocalhostBroker(config, **kwargs)
 
     redis = get_redis(config)
 
-    redis_broker = RedisBroker(redis)
+    redis_broker = RedisBroker(redis, config)
 
     return redis_broker
 
@@ -836,7 +836,7 @@ def get_uwsgi_broker(api_key, **kwargs):
     api_key, config, _, _ = _init_config(api_key, **kwargs)
 
     if api_key == 'localhost':
-        return LocalhostBroker(**kwargs)
+        return LocalhostBroker(config, **kwargs)
 
     uwsgi = get_uwsgi()
     uwsgi_broker = UWSGIBroker(uwsgi, config)

splitio/clients.py

@@ -6,11 +6,12 @@
 import time
 from splitio.treatments import CONTROL
 from splitio.splitters import Splitter
-from splitio.impressions import Impression, Label
+from splitio.impressions import Impression, Label, ImpressionListenerException
 from splitio.metrics import SDK_GET_TREATMENT
 from splitio.splits import ConditionType
 from splitio.events import Event
 
+
 class Key(object):
     def __init__(self, matching_key, bucketing_key):
         """Bucketing Key implementation"""
@@ -19,7 +20,7 @@ def __init__(self, matching_key, bucketing_key):
 
 
 class Client(object):
-    def __init__(self, broker, labels_enabled=True):
+    def __init__(self, broker, labels_enabled=True, impression_listener=None):
         """Basic interface of a Client. Specific implementations need to override the
         get_split_fetcher method (and optionally the get_splitter method).
         """
@@ -28,6 +29,7 @@ def __init__(self, broker, labels_enabled=True):
         self._broker = broker
         self._labels_enabled = labels_enabled
         self._destroyed = False
+        self._impression_listener = impression_listener
 
     @staticmethod
     def _get_keys(key):
@@ -49,6 +51,17 @@ def destroy(self):
         self._destroyed = True
         self._broker.destroy()
 
+    def _handle_custom_impression(self, impression, attributes):
+        '''
+        Handles custom impression if is present. Basically, sends the data
+        to client if some logic is wanted to do.
+        '''
+        if self._impression_listener is not None:
+            try:
+                self._impression_listener.log_impression(impression, attributes)
+            except ImpressionListenerException as e:
+                self._logger.exception(e)
+
     def get_treatment(self, key, feature, attributes=None):
         """
         Get the treatment for a feature and key, with an optional dictionary of attributes. This
@@ -106,6 +119,9 @@ def get_treatment(self, key, feature, attributes=None):
             impression = self._build_impression(matching_key, feature, _treatment, label,
                                                 _change_number, bucketing_key, start)
             self._record_stats(impression, start, SDK_GET_TREATMENT)
+
+            self._handle_custom_impression(impression, attributes)
+
             return _treatment
         except:
             self._logger.exception('Exception caught getting treatment for feature')
@@ -114,6 +130,8 @@ def get_treatment(self, key, feature, attributes=None):
                 impression = self._build_impression(matching_key, feature, CONTROL, Label.EXCEPTION,
                                                     self._broker.get_change_number(), bucketing_key, start)
                 self._record_stats(impression, start, SDK_GET_TREATMENT)
+
+                self._handle_custom_impression(impression, attributes)
             except:
                 self._logger.exception('Exception reporting impression into get_treatment exception block')
 
@@ -200,6 +218,7 @@ def track(self, key, traffic_type, event_type, value=None):
         )
         return self._broker.get_events_log().log_event(e)
 
+
 class MatcherClient(Client):
     """
     """

splitio/factories.py

@@ -5,6 +5,7 @@
 from splitio.brokers import get_self_refreshing_broker, get_redis_broker, get_uwsgi_broker
 from splitio.managers import RedisSplitManager, SelfRefreshingSplitManager, \
     LocalhostSplitManager, UWSGISplitManager
+from splitio.impressions import ImpressionListenerWrapper
 
 import logging
 
@@ -40,18 +41,21 @@ def __init__(self, api_key, **kwargs):
             config = kwargs['config']
 
         labels_enabled = config.get('labelsEnabled', True)
+
+        impression_listener = ImpressionListenerWrapper(config.get('impressionListener')) if 'impressionListener' in config else None # noqa: E501,E261
+
         if 'redisHost' in config or 'redisSentinels' in config:
             broker = get_redis_broker(api_key, **kwargs)
-            self._client = Client(broker, labels_enabled)
+            self._client = Client(broker, labels_enabled, impression_listener)
             self._manager = RedisSplitManager(broker)
         else:
             if 'uwsgiClient' in config and config['uwsgiClient']:
                 broker = get_uwsgi_broker(api_key, **kwargs)
-                self._client = Client(broker, labels_enabled)
+                self._client = Client(broker, labels_enabled, impression_listener)
                 self._manager = UWSGISplitManager(broker)
             else:
                 broker = get_self_refreshing_broker(api_key, **kwargs)
-                self._client = Client(broker, labels_enabled)
+                self._client = Client(broker, labels_enabled, impression_listener)
                 self._manager = SelfRefreshingSplitManager(broker)
 
     def client(self):  # pragma: no cover