Add redis storage backend

celery_dedupe/storage/__init__.py

@@ -0,0 +1,3 @@
+from .base import Storage
+
+__all__ = ['Storage']

celery_dedupe/storage/base.py

@@ -0,0 +1,18 @@
+class Storage(object):
+    '''
+    Defines the public api for Storages
+    '''
+
+    default_config = {}
+
+    def __init__(self, connection, **config):
+        self.connection = connection
+        self.config = self.default_config.copy()
+        self.config.update(config)
+
+    def obtain_lock(self, key, value):
+        '''
+        Responsible for obtaining a lock for the task
+        :returns boolean: Whether lock could be obtained or not
+        '''
+        raise NotImplementedError

celery_dedupe/storage/redis.py

@@ -0,0 +1,21 @@
+from . import Storage
+
+class RedisStorage(Storage):
+    '''
+    Storage class for redis.
+
+    Config:
+        expiry: number of seconds before expiring task, 0 to disable (default: 0)
+    '''
+
+    default_config = {
+        'expiry': 0,
+    }
+
+    def obtain_lock(self, key, value):
+        obtained = self.connection.setnx(key, value)
+        if self.config['expiry'] and obtained:
+            self.connection.expire(key, self.config['expiry'])
+
+        return obtained
+

dev-requirements.txt

@@ -0,0 +1,3 @@
+exam==0.10.5
+pytest==2.7.2
+redis==2.10.3

tests/test_storages.py

@@ -0,0 +1,30 @@
+from uuid import uuid4
+from redis import StrictRedis
+
+from celery_dedupe.storage.redis import RedisStorage
+
+class TestRedisStorage(object):
+
+    @property
+    def redis(self):
+        if hasattr(self, '_redis'):
+            return self._redis
+
+        self._redis = StrictRedis()
+        return self._redis
+
+    def test_lock_obtained(self):
+        storage = RedisStorage(self.redis)
+        assert storage.obtain_lock(uuid4(), '1')
+
+    def test_lock_obtained_with_expiry(self):
+        key = uuid4()
+        storage = RedisStorage(self.redis, expiry=10)
+        assert storage.obtain_lock(key, '1')
+        assert self.redis.ttl(key) == 10
+
+    def test_already_locked(self):
+        key = uuid4()
+        self.redis.setex(key, '1', 10)
+        storage = RedisStorage(self.redis, expiry=10)
+        assert not storage.obtain_lock(key, '1')