Merge pull request #333 from splitio/pluggable-sender-adapter

polishing

Author: chillaq

splitio/engine/impressions/adapters.py

@@ -5,6 +5,10 @@
 from splitio.storage.adapters.redis import RedisAdapterException
 
 _LOGGER = logging.getLogger(__name__)
+_MTK_QUEUE_KEY = 'SPLITIO.uniquekeys'
+_MTK_KEY_DEFAULT_TTL = 3600
+_IMP_COUNT_QUEUE_KEY = 'SPLITIO.impressions.count'
+_IMP_COUNT_KEY_DEFAULT_TTL = 3600
 
 class ImpressionsSenderAdapter(object, metaclass=abc.ABCMeta):
     """Impressions Sender Adapter interface."""
@@ -53,11 +57,6 @@ def _uniques_formatter(self, uniques):
 class RedisSenderAdapter(ImpressionsSenderAdapter):
     """In Memory Impressions Sender Adapter class."""
 
-    MTK_QUEUE_KEY = 'SPLITIO.uniquekeys'
-    MTK_KEY_DEFAULT_TTL = 3600
-    IMP_COUNT_QUEUE_KEY = 'SPLITIO.impressions.count'
-    IMP_COUNT_KEY_DEFAULT_TTL = 3600
-
     def __init__(self, redis_client):
         """
         Initialize In memory sender adapter instance
@@ -76,8 +75,8 @@ def record_unique_keys(self, uniques):
         """
         bulk_mtks = _uniques_formatter(uniques)
         try:
-            inserted = self._redis_client.rpush(self.MTK_QUEUE_KEY, *bulk_mtks)
-            self._expire_keys(self.MTK_QUEUE_KEY, self.MTK_KEY_DEFAULT_TTL, inserted, len(bulk_mtks))
+            inserted = self._redis_client.rpush(_MTK_QUEUE_KEY, *bulk_mtks)
+            self._expire_keys(_MTK_QUEUE_KEY, _MTK_KEY_DEFAULT_TTL, inserted, len(bulk_mtks))
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add mtks to redis')
@@ -96,11 +95,11 @@ def flush_counters(self, to_send):
             counted = 0
             pipe = self._redis_client.pipeline()
             for pf_count in to_send:
-                pipe.hincrby(self.IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
+                pipe.hincrby(_IMP_COUNT_QUEUE_KEY, pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
                 counted += pf_count.count
             resulted = sum(pipe.execute())
-            self._expire_keys(self.IMP_COUNT_QUEUE_KEY,
-                              self.IMP_COUNT_KEY_DEFAULT_TTL, resulted, counted)
+            self._expire_keys(_IMP_COUNT_QUEUE_KEY,
+                              _IMP_COUNT_KEY_DEFAULT_TTL, resulted, counted)
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add counters to redis')
@@ -122,11 +121,6 @@ def _expire_keys(self, queue_key, key_default_ttl, total_keys, inserted):
 class PluggableSenderAdapter(ImpressionsSenderAdapter):
     """In Memory Impressions Sender Adapter class."""
 
-    MTK_QUEUE_KEY = 'SPLITIO.uniquekeys'
-    MTK_KEY_DEFAULT_TTL = 3600
-    IMP_COUNT_QUEUE_KEY = 'SPLITIO.impressions.count'
-    IMP_COUNT_KEY_DEFAULT_TTL = 3600
-
     def __init__(self, adapter_client, prefix=None):
         """
         Initialize pluggable sender adapter instance
@@ -148,8 +142,8 @@ def record_unique_keys(self, uniques):
         """
         bulk_mtks = _uniques_formatter(uniques)
         try:
-            inserted = self._adapter_client.push_items(self.MTK_QUEUE_KEY, *bulk_mtks)
-            self._expire_keys(self._prefix + self.MTK_QUEUE_KEY, self.MTK_KEY_DEFAULT_TTL, inserted, len(bulk_mtks))
+            inserted = self._adapter_client.push_items(_MTK_QUEUE_KEY, *bulk_mtks)
+            self._expire_keys(self._prefix + _MTK_QUEUE_KEY, _MTK_KEY_DEFAULT_TTL, inserted, len(bulk_mtks))
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add mtks to storage adapter')
@@ -166,9 +160,9 @@ def flush_counters(self, to_send):
         try:
             resulted = 0
             for pf_count in to_send:
-                resulted = self._adapter_client.increment(self._prefix + self.IMP_COUNT_QUEUE_KEY + "." + pf_count.feature + "::" + str(pf_count.timeframe), pf_count.count)
-                self._expire_keys(self._prefix + self.IMP_COUNT_QUEUE_KEY + "." + pf_count.feature + "::" + str(pf_count.timeframe),
-                              self.IMP_COUNT_KEY_DEFAULT_TTL, resulted, pf_count.count)
+                key = self._prefix + _IMP_COUNT_QUEUE_KEY + "." + pf_count.feature + "::" + str(pf_count.timeframe)
+                resulted = self._adapter_client.increment(key, pf_count.count)
+                self._expire_keys(key, _IMP_COUNT_KEY_DEFAULT_TTL, resulted, pf_count.count)
             return True
         except RedisAdapterException:
             _LOGGER.error('Something went wrong when trying to add counters to storage adapter')

tests/engine/test_send_adapters.py

@@ -120,13 +120,13 @@ def test_record_unique_keys(self, mocker):
         ]
 
         sender_adapter.record_unique_keys(uniques)
-        assert(sorted(json.loads(adapter._keys[sender_adapter.MTK_QUEUE_KEY][0])["ks"]) == sorted(json.loads(formatted[0])["ks"]))
-        assert(sorted(json.loads(adapter._keys[sender_adapter.MTK_QUEUE_KEY][1])["ks"]) == sorted(json.loads(formatted[1])["ks"]))
-        assert(json.loads(adapter._keys[sender_adapter.MTK_QUEUE_KEY][0])["f"] == "feature1")
-        assert(json.loads(adapter._keys[sender_adapter.MTK_QUEUE_KEY][1])["f"] == "feature2")
-        assert(adapter._expire[sender_adapter.MTK_QUEUE_KEY] == sender_adapter.MTK_KEY_DEFAULT_TTL)
+        assert(sorted(json.loads(adapter._keys[adapters._MTK_QUEUE_KEY][0])["ks"]) == sorted(json.loads(formatted[0])["ks"]))
+        assert(sorted(json.loads(adapter._keys[adapters._MTK_QUEUE_KEY][1])["ks"]) == sorted(json.loads(formatted[1])["ks"]))
+        assert(json.loads(adapter._keys[adapters._MTK_QUEUE_KEY][0])["f"] == "feature1")
+        assert(json.loads(adapter._keys[adapters._MTK_QUEUE_KEY][1])["f"] == "feature2")
+        assert(adapter._expire[adapters._MTK_QUEUE_KEY] == adapters._MTK_KEY_DEFAULT_TTL)
         sender_adapter.record_unique_keys(uniques)
-        assert(adapter._expire[sender_adapter.MTK_QUEUE_KEY] != -1)
+        assert(adapter._expire[adapters._MTK_QUEUE_KEY] != -1)
 
     def test_flush_counters(self, mocker):
         """Test sending counters."""
@@ -139,8 +139,8 @@ def test_flush_counters(self, mocker):
         ]
 
         sender_adapter.flush_counters(counters)
-        assert(adapter._keys[sender_adapter.IMP_COUNT_QUEUE_KEY + "." + 'f1::123'] == 2)
-        assert(adapter._keys[sender_adapter.IMP_COUNT_QUEUE_KEY + "." + 'f2::123'] == 123)
-        assert(adapter._expire[sender_adapter.IMP_COUNT_QUEUE_KEY + "." + 'f1::123'] == sender_adapter.IMP_COUNT_KEY_DEFAULT_TTL)
+        assert(adapter._keys[adapters._IMP_COUNT_QUEUE_KEY + "." + 'f1::123'] == 2)
+        assert(adapter._keys[adapters._IMP_COUNT_QUEUE_KEY + "." + 'f2::123'] == 123)
+        assert(adapter._expire[adapters._IMP_COUNT_QUEUE_KEY + "." + 'f1::123'] == adapters._IMP_COUNT_KEY_DEFAULT_TTL)
         sender_adapter.flush_counters(counters)
-        assert(adapter._expire[sender_adapter.IMP_COUNT_QUEUE_KEY + "." + 'f2::123'] == sender_adapter.IMP_COUNT_KEY_DEFAULT_TTL)
+        assert(adapter._expire[adapters._IMP_COUNT_QUEUE_KEY + "." + 'f2::123'] == adapters._IMP_COUNT_KEY_DEFAULT_TTL)