Merge branch 'master' of [email protected]:fakechris/pyres

Author: fakechris

src/pyres/__init__.py

@@ -24,26 +24,26 @@ def __init__(self, server="localhost:6379"):
 
     def push(self, queue, item):
         self.watch_queue(queue)
-        self.redis.push("queue:%s" % queue, ResQ.encode(item))
+        self.redis.push("resque:queue:%s" % queue, ResQ.encode(item))
 
     def pop(self, queue):
-        ret = self.redis.pop("queue:%s" % queue)
+        ret = self.redis.pop("resque: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("resque:queue:%s" % queue))
 
     def watch_queue(self, queue):
         if queue in self._watched_queues:
             return
         else:
-            if self.redis.sadd('queues',str(queue)):
+            if self.redis.sadd('resque:queues',str(queue)):
                 self._watched_queues.add(queue)
 
     def peek(self, queue, start=0, count=1):
-        return self.list_range('queue:%s' % queue, start, count)
+        return self.list_range('resque:queue:%s' % queue, start, count)
 
     def list_range(self, key, start, count):
         items = self.redis.lrange(key,start,start+count-1)
@@ -79,7 +79,7 @@ def enqueue_from_string(self, klass_as_string, queue, *args):
         self.push(queue, {'class':klass_as_string,'args':args})
 
     def queues(self):
-        return self.redis.smembers("queues")
+        return self.redis.smembers("resque:queues")
 
     def info(self):
         pending = 0
@@ -116,8 +116,8 @@ def working(self):
     def remove_queue(self, queue):
         if queue in self._watched_queues:
             self._watched_queues.remove(queue)
-        self.redis.srem('queues',queue)
-        del self.redis['queue:%s' % queue]
+        self.redis.srem('resque:queues',queue)
+        del self.redis['resque:queue:%s' % queue]
 
     @classmethod
     def encode(cls, item):
@@ -144,7 +144,7 @@ def _enqueue(cls, klass, *args):
 class Stat(object):
     def __init__(self, name, resq):
         self.name = name
-        self.key = "stat:%s" % self.name
+        self.key = "resque:stat:%s" % self.name
         self.resq = resq
 
     def get(self):

src/pyres/extensions.py

@@ -3,6 +3,8 @@
 except:
     import sys
     sys.exit("multiprocessing was not available")
+import os, datetime, time, signal
+from pyres import ResQ
 
 from pyres.exceptions import NoQueueError
 from pyres.worker import Worker

src/pyres/worker.py

@@ -6,6 +6,7 @@
 import os, sys
 import time
 import simplejson
+
 class Worker(object):
     def __init__(self, queues=[], server="localhost:6379"):
         self.queues = queues
@@ -25,13 +26,13 @@ def validate_queues(self):
             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('resque: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('resque:workers',str(self))
 
     def startup(self):
         self.register_signal_handlers()
@@ -108,11 +109,11 @@ def working_on(self, job):
             'payload': job._payload
         }
         data = simplejson.dumps(data)
-        self.resq.redis["worker:%s" % str(self)] = data
+        self.resq.redis["resque:worker:%s" % str(self)] = data
 
     def done_working(self):
         self.processed()
-        self.resq.redis.delete("worker:%s" % str(self))
+        self.resq.redis.delete("resque:worker:%s" % str(self))
 
     def processed(self):
         total_processed = Stat("processed", self.resq)
@@ -127,7 +128,7 @@ def failed(self):
         stat.incr()
 
     def job(self):
-        data = self.resq.redis.get("worker:%s" % self)
+        data = self.resq.redis.get("resque:worker:%s" % self)
         if data:
             return ResQ.decode(data)
         return {}
@@ -151,7 +152,7 @@ def all(cls, host="localhost:6379"):
             resq = ResQ(host)
         elif isinstance(host, ResQ):
             resq = host
-        return resq.redis.smembers('workers')
+        return resq.redis.smembers('resque:workers')
 
     @classmethod
     def working(cls, host):
@@ -162,7 +163,7 @@ def working(cls, host):
         total = []
         for key in Worker.all(host):
             if Worker.exists(key, resq):
-                total.append('worker:%s' % key)
+                total.append('resque:worker:%s' % key)
         names = []
         for key in total:
             value = resq.redis.get(key)
@@ -183,7 +184,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('resque:workers', worker_id)
 
 
 if __name__ == "__main__":

src/resweb/server.py

@@ -9,8 +9,11 @@
 
 @get("/")
 def index(request):
-    queues = ResQ.queues(HOST)
-    failure_count = Failure.count(ResQ(HOST))
+    #queues = ResQ.queues(HOST)
+    #failure_count = Failure.count(ResQ(HOST))
+    rq = ResQ(HOST)
+    queues = rq.queues()
+    failure_count = Failure.count(rq)
     template = env.get_template('overview.html')
     dic = {
         'queues':queues,
@@ -33,9 +36,12 @@ def working(request):
     return str(template.render(dic))
 
 @get("/queues/")
-def queues(request):
-    queues = ResQ.queues(HOST)
-    failure_count = Failure.count(ResQ(HOST))
+def queues(request):    
+    #queues = ResQ.queues(HOST)
+    #failure_count = Failure.count(ResQ(HOST))
+    rq = ResQ(HOST)
+    queues = rq.queues()
+    failure_count = Failure.count(rq)
     template = env.get_template('queues.html')
     dic = {
         'queues':queues,