Updated the ResQ class to better reflect resque

Also added __getattr__ to the stats class and
a property for the redis object in ResQ

Author: Matt George

src/pyres/__init__.py

@@ -17,88 +17,142 @@ def str_to_class(s):
         return None
 
 class ResQ(object):
-    _res = None
+    
     def __init__(self, server="localhost:6379"):
-        self._server = server
-        host, port = server.split(':')
-        self._redis = Redis(host=host, port=int(port))
+        self.redis = server
         self._watched_queues = {}
-    
+
     def push(self, queue, item):
         self.watch_queue(queue)
-        self._redis.push("queue:%s" % queue, ResQ.encode(item))
-    
+        self.redis.push("queue:%s" % queue, ResQ.encode(item))
+
     def pop(self, queue):
-        ret = self._redis.pop("queue:%s" % queue)
+        ret = self.redis.pop("queue:%s" % queue)
         if ret:
             return ResQ.decode(ret)
         return ret
-    
+
     def size(self, queue):
-        return int(self._redis.llen("queue:%s" % queue))
-    
+        return int(self.redis.llen("queue:%s" % queue))
+
     def watch_queue(self, queue):
         if self._watched_queues.has_key(queue):
             return
         else:
-            if self._redis.sadd('queues',str(queue)):
+            if self.redis.sadd('queues',str(queue)):
                 self._watched_queues[queue] = queue
+
     def peek(self, queue, start=0, count=1):
         return self.list_range('queue:%s' % queue, start, count)
-    
+
     def list_range(self, key, start, count):
-        items = self._redis.lrange(key,start,start+count-1)
+        items = self.redis.lrange(key,start,start+count-1)
         ret_list = []
         for i in items:
             ret_list.append(ResQ.decode(i))
         return ret_list
+
+    def _get_redis(self):
+        return self._redis
+
+    def _set_redis(self, server):
+        if isinstance(server, basestring):
+            self.dsn = server
+            host, port = server.split(':')
+            self._redis = Redis(host=host, port=int(port))
+        elif isinstance(server, Redis):
+            self.dsn = '%s:%s' % (server.host,server.port)
+            self._redis = server
+        else:
+            raise Exception("I don't know what to do with %s" % str(server))
+    redis = property(_get_redis, _set_redis)
+
+    def enqueue(self, klass, *args):
+        queue = getattr(klass,'queue', None)
+        #print cls._res
+        if queue:
+            class_name = '%s.%s' % (klass.__module__, klass.__name__)
+            #print class_name
+            self.push(queue, {'class':class_name,'args':args})
+            #Job.create(queue, klass,*args)
+
+    def queues(self):
+        return self.redis.smembers("queues")
+    
+    def info(self):
+        pending = 0
+        for q in self.queues():
+             pending += self.size(q)
+        return {
+            'pending'   : pending,
+            'processed' : Stat['processed'],
+            'queues'    : queues.size,
+            'workers'   : len(self.workers),
+            'working'   : len(self.working),
+            'failed'    : Stat['failed'],
+            'servers'   : [redis.server]
+        }
+    
+    def keys(self):
+        raise NotImplementedError
+    
+    def reserve(self, queue):
+        from pyres.job import Job
+        return Job.reserve(queue, self)
+    
+    def __str__(self):
+        return "PyRes Client connected to %s" % self.redis.server
+    
+    def workers(self):
+        raise NotImplementedError
+    
+    def working(self):
+        raise NotImplementedError
+    
     @classmethod
     def encode(cls, item):
         return simplejson.dumps(item)
-    
+
     @classmethod
     def decode(cls, item):
         if item:
             ret = simplejson.loads(item)
             return ret
         return None
 
-    @classmethod
-    def enqueue(cls, klass, *args):
+    @staticmethod
+    def enqueue(klass, *args):
         queue = getattr(klass,'queue', None)
         #print cls._res
-        resq = cls()
+        _self = ResQ()
         if queue:
             class_name = '%s.%s' % (klass.__module__, klass.__name__)
             #print class_name
-            resq.push(queue, {'class':class_name,'args':args})
+            _self.push(queue, {'class':class_name,'args':args})
             #Job.create(queue, klass,*args)
-    
-    @classmethod
-    def queues(cls, server="localhost:6379"):
-        resq = cls(server)
-        return resq._redis.smembers("queues")
-    
-    #@classmethod
-    #def working(cls, server="localhost:6379"):
-    #    resq = cls(server)
-    #    return Worker.working(resq)
+
 class Stat(object):
     def __init__(self, name, resq):
         self.name = name
         self.key = "stat:%s" % self.name
         self.resq = resq
 
+    def __getitem__(self, name):
+        val = self.get(name)
+        if val is None:
+            raise KeyError
+        return val
+    
     def get(self):
-        val = self.resq._redis.get(self.key)
+        val = self.resq.redis.get(self.key)
         return int(val)
 
     def incr(self, ammount=1):
-        self.resq._redis.incr(self.key, ammount)
+        self.resq.redis.incr(self.key, ammount)
 
     def decr(self, ammount=1):
-        self.resq._redis.decr(self.key, ammount)
+        self.resq.redis.decr(self.key, ammount)
 
     def clear(self):
-        self.resq._redis.delete(self.key)
+        self.resq.redis.delete(self.key)
 

src/pyres/failure.py

@@ -15,11 +15,11 @@ def save(self, resq):
             'queue'     : self._queue
         }
         data = ResQ.encode(data)
-        resq._redis.push('failed', data)
+        resq.redis.push('failed', data)
 
     @classmethod
     def count(cls, resq):
-        return int(resq._redis.llen('failed'))
+        return int(resq.redis.llen('failed'))
 
     @classmethod
     def create(cls, options={}):

src/pyres/job.py

@@ -23,6 +23,5 @@ def fail(self, exception):
     @classmethod
     def reserve(cls, queue, res):
         payload = res.pop(queue)
-        print "payload: %s" % payload
         if payload:
             return cls(queue, payload, res)

src/pyres/manager.py

@@ -26,10 +26,10 @@ def startup(self):
         self.register_signals()
 
     def register_manager(self):
-        self.resq._redis.sadd('managers',str(self))
+        self.resq.redis.sadd('managers',str(self))
 
     def unregister_manager(self):
-        self.resq._redis.srem('managers',str(self))
+        self.resq.redis.srem('managers',str(self))
 
     def register_signals(self):
         signal.signal(signal.SIGTERM, self.shutdown_all)

src/pyres/worker.py

@@ -7,26 +7,31 @@
 import time
 import simplejson
 class Worker(object):
-    def __init__(self, queues=[], host="localhost:6379"):
+    def __init__(self, queues=[], server="localhost:6379"):
         self.queues = queues
         self.validate_queues()
         self._shutdown = False
         self.child = None
-        self.resq = ResQ(host)
+        if isinstance(server,basestring):
+            self.resq = ResQ(server)
+        elif isinstance(server, ResQ):
+            self.resq = server
+        else:
+            raise Exception("Bad server argument")
+        
 
     def validate_queues(self):
         if not self.queues:
             raise NoQueueError("Please give each worker at least one queue.")
 
     def register_worker(self):
-        
-        self.resq._redis.sadd('workers',str(self))
+        self.resq.redis.sadd('workers',str(self))
         #self.resq._redis.add("worker:#{self}:started", Time.now.to_s)
         #Stat.clear("processed:#{self}")
         #Stat.clear("failed:#{self}")
 
     def unregister_worker(self):
-        self.resq._redis.srem('workers',str(self))
+        self.resq.redis.srem('workers',str(self))
 
     def startup(self):
         self.register_signal_handlers()
@@ -103,14 +108,14 @@ def working_on(self, job):
             'payload': job._payload
         }
         data = simplejson.dumps(data)
-        self.resq._redis.set("worker:%s" % str(self), data)
+        self.resq.redis["worker:%s" % str(self)] = data
 
     def job(self):
-        return ResQ.decode(self.resq._redis.get("worker:%s" % self)) or {}
+        return ResQ.decode(self.resq.redis.get("worker:%s" % self)) or {}
 
     def done_working(self):
         self.processed()
-        self.resq._redis.delete("worker:%s" % str(self))
+        self.resq.redis.delete("worker:%s" % str(self))
 
     def processed(self):
         total_processed = Stat("processed", self.resq)
@@ -125,7 +130,7 @@ def failed(self):
         stat.incr()
 
     def job(self):
-        data = self.resq._redis.get("worker:%s" % self)
+        data = self.resq.redis.get("worker:%s" % self)
         if data:
             return ResQ.decode(data)
         return {}
@@ -139,13 +144,15 @@ def __str__(self):
         return '%s:%s:%s' % (hostname, pid, ','.join(self.queues))
 
     @classmethod
-    def run(cls, queues, host):
-        worker = cls(queues=queues, host=host)
+    def run(cls, queues, server):
+        worker = cls(queues=queues, host=server)
         worker.work()
+    
     @classmethod
     def all(cls, host):
         resq = ResQ(host)
-        return resq._redis.smembers('workers')
+        return resq.redis.smembers('workers')
+    
     @classmethod
     def working(cls, host):
         resq = ResQ(host)
@@ -169,7 +176,7 @@ def find(cls, worker_id, resq):
 
     @classmethod
     def exists(cls, worker_id, resq):
-        return resq._redis.sismember('workers', worker_id)
+        return resq.redis.sismember('workers', worker_id)
 
 class JuniorWorker(Worker):
     def work(self, interval=5):

src/tests.py

@@ -8,7 +8,9 @@ class Basic(object):
 
     @staticmethod
     def perform(name):
-        return "name:%s" % name
+        s = "name:%s" % name
+        print s
+        return s
 
 def test_str_to_class():
     ret = str_to_class('tests.Basic')
@@ -17,19 +19,21 @@ def test_str_to_class():
 class PyResTests(unittest.TestCase):
     def setUp(self):
         self.resq = ResQ()
-        self.redis = self.resq._redis
+        self.redis = self.resq.redis
         self.redis.flush(True)
 
     def tearDown(self):
+        self.redis.flush(True)
         del self.redis
         del self.resq
 
 
 class ResQTests(PyResTests):
     def test_enqueue(self):
-        ResQ.enqueue(Basic,"test1")
-        ResQ.enqueue(Basic,"test2")
-        assert self.redis.llen("queue:basic") == 2
+        self.resq.enqueue(Basic,"test1")
+        self.resq.enqueue(Basic,"test2")
+        ResQ.enqueue(Basic, "test3")
+        assert self.redis.llen("queue:basic") == 3
         assert self.redis.sismember('queues','basic')
 
     def test_push(self):
@@ -49,24 +53,24 @@ def test_pop(self):
         assert self.redis.llen('queue:pushq') == 0
 
     def test_peek(self):
-        ResQ.enqueue(Basic,"test1")
-        ResQ.enqueue(Basic,"test2")
+        self.resq.enqueue(Basic,"test1")
+        self.resq.enqueue(Basic,"test2")
         assert len(self.resq.peek('basic',0,20)) == 2
 
 class JobTests(PyResTests):
     def test_reserve(self):
-        ResQ.enqueue(Basic,"test1")
+        self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic', self.resq)
         assert job._queue == 'basic'
         assert job._payload
 
     def test_perform(self):
-        ResQ.enqueue(Basic,"test1")
+        self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
         assert job.perform() == "name:test1"
 
     def test_fail(self):
-        ResQ.enqueue(Basic,"test1")
+        self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
         assert self.redis.llen('failed') == 0
         job.fail("problem")
@@ -96,7 +100,7 @@ def test_unregister(self):
 
     def test_working_on(self):
         name = "%s:%s:%s" % (os.uname()[1],os.getpid(),'basic')
-        ResQ.enqueue(Basic,"test1")
+        self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic', self.resq)
         worker = Worker(['basic'])
         worker.working_on(job)
@@ -127,5 +131,12 @@ def test_failed(self):
         assert self.redis.get("stat:failed:%s" % name) == 2
 
     def test_process(self):
-        assert False
+        name = "%s:%s:%s" % (os.uname()[1],os.getpid(),'basic')
+        self.resq.enqueue(Basic,"test1")
+        job = Job.reserve('basic', self.resq)
+        worker = Worker(['basic'])
+        worker.process(job)
+        assert not self.redis.get('worker:%s' % worker)
+        assert not self.redis.get("stat:failed")
+        assert not self.redis.get("stat:failed:%s" % name)