improvements

Author: Matias Melograno

splitio/client/factory.py

@@ -354,7 +354,7 @@ def _build_redis_factory(api_key, cfg):
         'telemetry': RedisTelemetryStorage(redis_adapter, sdk_metadata)
     }
     recorder = PipelinedRecorder(
-        redis_adapter,
+        redis_adapter.pipeline(),
         ImpressionsManager(cfg['impressionsMode'], False,
                            _wrap_impression_listener(cfg['impressionListener'], sdk_metadata)),
         storages['telemetry'],

splitio/recorder/recorder.py

@@ -91,12 +91,12 @@ def record_track_stats(self, event):
 class PipelinedRecorder(StatsRecorder):
     """PipelinedRecorder class."""
 
-    def __init__(self, redis, impressions_manager, telemetry_storage, event_storage, impression_storage):
+    def __init__(self, pipe, impressions_manager, telemetry_storage, event_storage, impression_storage):
         """
         Class constructor.
 
-        :param redis: redis client instance
-        :type redis: splitio.storage.adapters.redis.RedisAdapter
+        :param pipe: redis pipeline
+        :type pipe: splitio.storage.adapters.redis.RedisPipelineAdapter
         :param impressions_manager: impression manager instance
         :type impressions_manager: splitio.engine.impressions.Manager
         :param telemetry_storage: telemetry storage instance
@@ -106,7 +106,7 @@ def __init__(self, redis, impressions_manager, telemetry_storage, event_storage,
         :param impression_storage: impression storage instance
         :type impression_storage: splitio.storage.redis.RedisImpressionsStorage
         """
-        self._redis = redis
+        self._pipe = pipe
         self._impressions_manager = impressions_manager
         self._telemetry_storage = telemetry_storage
         self._event_sotrage = event_storage
@@ -125,10 +125,9 @@ def record_treatment_stats(self, impressions, latency, operation):
         """
         try:
             impressions = self._impressions_manager.process_impressions(impressions)
-            pipe = self._redis.pipeline()
-            self._impression_storage.add_impressions_to_pipe(impressions, pipe)
-            self._telemetry_storage.add_latency_to_pipe(operation, latency, pipe)
-            result = pipe.execute()
+            self._impression_storage.add_impressions_to_pipe(impressions, self._pipe)
+            self._telemetry_storage.add_latency_to_pipe(operation, latency, self._pipe)
+            result = self._pipe.execute()
             if len(result) == 2:
                 self._impression_storage.expire_key(result[0], len(impressions))
         except Exception:  # pylint: disable=broad-except

splitio/storage/adapters/redis.py

@@ -53,8 +53,8 @@ class SentinelConfigurationException(Exception):
     pass
 
 
-class PrefixTrait(object):
-    """Prefix generator."""
+class PrefixHelper(object):
+    """PrefixHelper generator."""
 
     def __init__(self, prefix=None):
         """
@@ -137,7 +137,7 @@ def __init__(self, decorated, prefix=None):
         :param prefix: User prefix to add.
         """
         self._decorated = decorated
-        self._prefix_trait = PrefixTrait(prefix)
+        self._prefix_helper = PrefixHelper(prefix)
 
     # Below starts a list of methods that implement the interface of a standard
     # redis client.
@@ -147,7 +147,7 @@ def keys(self, pattern):
         try:
             return [
                 _bytes_to_string(key)
-                for key in self._prefix_trait.remove_prefix(self._decorated.keys(self._prefix_trait.add_prefix(pattern)))
+                for key in self._prefix_helper.remove_prefix(self._decorated.keys(self._prefix_helper.add_prefix(pattern)))
             ]
         except RedisError as exc:
             raise_from(RedisAdapterException('Failed to execute keys operation'), exc)
@@ -156,43 +156,43 @@ def set(self, name, value, *args, **kwargs):
         """Mimic original redis function but using user custom prefix."""
         try:
             return self._decorated.set(
-                self._prefix_trait.add_prefix(name), value, *args, **kwargs
+                self._prefix_helper.add_prefix(name), value, *args, **kwargs
             )
         except RedisError as exc:
-            raise RedisAdapterException('Failed to execute set operation', exc)
+            raise_from(RedisAdapterException('Failed to execute set operation'), exc)
 
     def get(self, name):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return _bytes_to_string(self._decorated.get(self._prefix_trait.add_prefix(name)))
+            return _bytes_to_string(self._decorated.get(self._prefix_helper.add_prefix(name)))
         except RedisError as exc:
-            raise RedisAdapterException('Error executing get operation', exc)
+            raise_from(RedisAdapterException('Error executing get operation'), exc)
 
     def setex(self, name, time, value):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.setex(self._prefix_trait.add_prefix(name), time, value)
+            return self._decorated.setex(self._prefix_helper.add_prefix(name), time, value)
         except RedisError as exc:
-            raise RedisAdapterException('Error executing setex operation', exc)
+            raise_from(RedisAdapterException('Error executing setex operation'), exc)
 
     def delete(self, *names):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.delete(*self._prefix_trait.add_prefix(list(names)))
+            return self._decorated.delete(*self._prefix_helper.add_prefix(list(names)))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing delete operation'), exc)
 
     def exists(self, name):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.exists(self._prefix_trait.add_prefix(name))
+            return self._decorated.exists(self._prefix_helper.add_prefix(name))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing exists operation'), exc)
 
     def lrange(self, key, start, end):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.lrange(self._prefix_trait.add_prefix(key), start, end)
+            return self._decorated.lrange(self._prefix_helper.add_prefix(key), start, end)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing exists operation'), exc)
 
@@ -201,7 +201,7 @@ def mget(self, names):
         try:
             return [
                 _bytes_to_string(item)
-                for item in self._decorated.mget(self._prefix_trait.add_prefix(names))
+                for item in self._decorated.mget(self._prefix_helper.add_prefix(names))
             ]
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing mget operation'), exc)
@@ -211,106 +211,106 @@ def smembers(self, name):
         try:
             return [
                 _bytes_to_string(item)
-                for item in self._decorated.smembers(self._prefix_trait.add_prefix(name))
+                for item in self._decorated.smembers(self._prefix_helper.add_prefix(name))
             ]
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing smembers operation'), exc)
 
     def sadd(self, name, *values):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.sadd(self._prefix_trait.add_prefix(name), *values)
+            return self._decorated.sadd(self._prefix_helper.add_prefix(name), *values)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing sadd operation'), exc)
 
     def srem(self, name, *values):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.srem(self._prefix_trait.add_prefix(name), *values)
+            return self._decorated.srem(self._prefix_helper.add_prefix(name), *values)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing srem operation'), exc)
 
     def sismember(self, name, value):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.sismember(self._prefix_trait.add_prefix(name), value)
+            return self._decorated.sismember(self._prefix_helper.add_prefix(name), value)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing sismember operation'), exc)
 
     def eval(self, script, number_of_keys, *keys):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.eval(script, number_of_keys, *self._prefix_trait.add_prefix(list(keys)))
+            return self._decorated.eval(script, number_of_keys, *self._prefix_helper.add_prefix(list(keys)))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing eval operation'), exc)
 
     def hset(self, name, key, value):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.hset(self._prefix_trait.add_prefix(name), key, value)
+            return self._decorated.hset(self._prefix_helper.add_prefix(name), key, value)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing hset operation'), exc)
 
     def hget(self, name, key):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return _bytes_to_string(self._decorated.hget(self._prefix_trait.add_prefix(name), key))
+            return _bytes_to_string(self._decorated.hget(self._prefix_helper.add_prefix(name), key))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing hget operation'), exc)
 
     def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.incr(self._prefix_trait.add_prefix(name), amount)
+            return self._decorated.incr(self._prefix_helper.add_prefix(name), amount)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing incr operation'), exc)
 
     def getset(self, name, value):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return _bytes_to_string(self._decorated.getset(self._prefix_trait.add_prefix(name), value))
+            return _bytes_to_string(self._decorated.getset(self._prefix_helper.add_prefix(name), value))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing getset operation'), exc)
 
     def rpush(self, key, *values):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.rpush(self._prefix_trait.add_prefix(key), *values)
+            return self._decorated.rpush(self._prefix_helper.add_prefix(key), *values)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing rpush operation'), exc)
 
     def expire(self, key, value):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.expire(self._prefix_trait.add_prefix(key), value)
+            return self._decorated.expire(self._prefix_helper.add_prefix(key), value)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing expire operation'), exc)
 
     def rpop(self, key):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return _bytes_to_string(self._decorated.rpop(self._prefix_trait.add_prefix(key)))
+            return _bytes_to_string(self._decorated.rpop(self._prefix_helper.add_prefix(key)))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing rpop operation'), exc)
 
     def ttl(self, key):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.ttl(self._prefix_trait.add_prefix(key))
+            return self._decorated.ttl(self._prefix_helper.add_prefix(key))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing ttl operation'), exc)
 
     def lpop(self, key):
         """Mimic original redis function but using user custom prefix."""
         try:
-            return self._decorated.lpop(self._prefix_trait.add_prefix(key))
+            return self._decorated.lpop(self._prefix_helper.add_prefix(key))
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing lpop operation'), exc)
 
     def pipeline(self):
         """Mimic original redis pipeline."""
         try:
-            return RedisPipelineAdapter(self._decorated, self._prefix_trait)
+            return RedisPipelineAdapter(self._decorated, self._prefix_helper)
         except RedisError as exc:
             raise_from(RedisAdapterException('Error executing ttl operation'), exc)
 
@@ -322,23 +322,23 @@ class RedisPipelineAdapter(object):
     Adds an extra layer handling addition/removal of user prefix when handling
     keys
     """
-    def __init__(self, decorated, prefix_trait):
+    def __init__(self, decorated, prefix_helper):
         """
         Store the user prefix and the redis client instance.
 
         :param decorated: Instance of redis cache client to decorate.
-        :param prefix_trait: Prefix Trait utility
+        :param _prefix_helper: PrefixHelper utility
         """
-        self._prefix_trait = prefix_trait
+        self._prefix_helper = prefix_helper
         self._pipe = decorated.pipeline()
 
     def rpush(self, key, *values):
         """Mimic original redis function but using user custom prefix."""
-        self._pipe.rpush(self._prefix_trait.add_prefix(key), *values)
+        self._pipe.rpush(self._prefix_helper.add_prefix(key), *values)
 
     def incr(self, name, amount=1):
         """Mimic original redis function but using user custom prefix."""
-        self._pipe.incr(self._prefix_trait.add_prefix(name), amount)
+        self._pipe.incr(self._prefix_helper.add_prefix(name), amount)
 
     def execute(self):
         """Mimic original redis function but using user custom prefix."""

tests/client/test_factory.py

@@ -20,7 +20,7 @@
 from splitio.sync.split import SplitSynchronizer
 from splitio.sync.segment import SegmentSynchronizer
 from splitio.recorder.recorder import PipelinedRecorder, StandardRecorder
-from splitio.storage.adapters.redis import RedisAdapter
+from splitio.storage.adapters.redis import RedisAdapter, RedisPipelineAdapter
 
 
 class SplitFactoryTests(object):
@@ -130,11 +130,11 @@ def test_redis_client_creation(self, mocker):
             ssl_cert_reqs='some_cert_req',
             ssl_ca_certs='some_ca_cert',
             max_connections=999
-        )]
+        ), mocker.call().pipeline()]
         assert factory._labels_enabled is False
         assert isinstance(factory._recorder, PipelinedRecorder)
         assert isinstance(factory._recorder._impressions_manager, ImpressionsManager)
-        assert isinstance(factory._recorder._redis, RedisAdapter)
+        assert isinstance(factory._recorder._pipe, RedisPipelineAdapter)
         assert isinstance(factory._recorder._telemetry_storage, redis.RedisTelemetryStorage)
         assert isinstance(factory._recorder._event_sotrage, redis.RedisEventsStorage)
         assert isinstance(factory._recorder._impression_storage, redis.RedisImpressionsStorage)

tests/storage/adapters/test_redis_adapter.py

@@ -189,8 +189,8 @@ def test_forwarding(self, mocker):
         redis_mock = mocker.Mock(StrictRedis)
         redis_mock_2 = mocker.Mock(Redis)
         redis_mock.pipeline.return_value = redis_mock_2
-        prefix_trait = redis.PrefixTrait('some_prefix')
-        adapter = redis.RedisPipelineAdapter(redis_mock, prefix_trait)
+        prefix_helper = redis.PrefixHelper('some_prefix')
+        adapter = redis.RedisPipelineAdapter(redis_mock, prefix_helper)
 
         adapter.rpush('key1', 'value1', 'value2')
         assert redis_mock_2.rpush.mock_calls[0] == mocker.call('some_prefix.key1', 'value1', 'value2')