Merge branch 'refactor_paths'

Author: Matt George

AUTHORS.md

@@ -0,0 +1,5 @@
+## Authors
+* Matt George
+* Chris Song
+
+Inspired by Resque, by Chris Wanstrath

HISTORY.md

@@ -1,3 +1,10 @@
+## 0.2.1 (2009-12-10)
+
+* updated setup.py
+* refactored package for better testing
+* resque namespacing by fakechris
+* smarter import from string by fakechris
+
 ## 0.2.0 (2009-12-09)
 
 * Better web interface via resweb

coverage.report

@@ -0,0 +1,11 @@
+Name               Stmts   Exec  Cover   Missing
+------------------------------------------------
+pyres                138    131    94%   26, 39, 98, 133-134, 144-145
+pyres.exceptions       2      2   100%   
+pyres.failure         23     22    95%   41
+pyres.job             23     23   100%   
+pyres.worker         189    146    77%   66, 74, 84-112, 161, 179, 186, 230-241
+------------------------------------------------
+TOTAL                375    324    86%   
+----------------------------------------------------------------------
+Ran 32 tests in 0.884s

src/pyres/__init__.py pyres/__init__.pysrc/pyres/__init__.py renamed to pyres/__init__.py

@@ -1,9 +1,31 @@
-__version__ = '0.2.0'
+__version__ = '0.2.1'
 
 from redis import Redis
 import simplejson
 
 import types
+
+def my_import(name):
+    mod = __import__(name)    
+    components = name.split('.')    
+    for comp in components[1:]:        
+        mod = getattr(mod, comp)    
+    return mod
+
+def safe_str_to_class(s):
+    lst = s.split(".")
+    klass = lst[-1]
+    mod_list = lst[:-1]
+    module = ".".join(mod_list)
+    try:
+        mod = my_import(module)
+        if hasattr(mod, klass):
+            return getattr(mod, klass)
+        else:
+            return None
+    except ImportError:
+        return None
+
 def str_to_class(s):
     lst = s.split(".")
     klass = lst[-1]
@@ -20,32 +42,37 @@ def str_to_class(s):
 
 class ResQ(object):
 
-    def __init__(self, server="localhost:6379"):
+    def __init__(self, server="localhost:6379", password=None, 
+                 timeout=None, retry_connection=True):
+        self.timeout = timeout
+        self.retry_connection = retry_connection
         self.redis = server
+        if password:
+            self.redis.auth(password)
         self._watched_queues = set()
 
     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)
@@ -61,7 +88,9 @@ def _set_redis(self, server):
         if isinstance(server, basestring):
             self.dsn = server
             host, port = server.split(':')
-            self._redis = Redis(host=host, port=int(port))
+            self._redis = Redis(host=host, port=int(port), 
+                                retry_connection=self.retry_connection,
+                                timeout=self.timeout)
         elif isinstance(server, Redis):
             self.dsn = '%s:%s' % (server.host,server.port)
             self._redis = server
@@ -81,7 +110,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
@@ -118,8 +147,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):
@@ -146,7 +175,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/exceptions.py pyres/exceptions.pysrc/pyres/exceptions.py renamed to pyres/exceptions.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/extensions.py pyres/extensions.pysrc/pyres/extensions.py renamed to pyres/extensions.py

@@ -2,12 +2,12 @@
 from pyres import ResQ 
 import sys, traceback
 class Failure(object):
-    def __init__(self, exp, queue, payload):
+    def __init__(self, exp, queue, payload, worker=None):
         excc, _, tb = sys.exc_info()
 
         self._exception = excc
         self._traceback = tb
-        #self._worker = worker
+        self._worker = worker
         self._queue = queue
         self._payload = payload
 
@@ -30,13 +30,14 @@ def save(self, resq):
             'payload'   : self._payload,
             'error'     : self._parse_message(self._exception),
             'backtrace' : self._parse_traceback(self._traceback),
-            'queue'     : self._queue
+            'queue'     : self._queue,
+            'worker'    : self._worker
         }
         data = ResQ.encode(data)
-        resq.redis.push('failed', data)
+        resq.redis.push('resque:failed', data)
 
     @classmethod
     def count(cls, resq):
-        return int(resq.redis.llen('failed'))
+        return int(resq.redis.llen('resque:failed'))
 
 

src/pyres/failure.py pyres/failure.pysrc/pyres/failure.py renamed to pyres/failure.py

@@ -1,14 +1,15 @@
-from pyres import ResQ, str_to_class
+from pyres import ResQ, str_to_class, safe_str_to_class
 from pyres.failure import Failure
 class Job(object):
-    def __init__(self, queue, payload, resq):
+    def __init__(self, queue, payload, resq, worker=None):
         self._queue = queue
         self._payload = payload
         self.resq = resq
+        self._worker = worker
 
     def perform(self):
         payload_class_str = self._payload["class"]
-        payload_class = str_to_class(payload_class_str)
+        payload_class = safe_str_to_class(payload_class_str)
         args = self._payload.get("args", None)
         if args:
             return payload_class.perform(*args)
@@ -17,11 +18,11 @@ def perform(self):
 
     def fail(self, exception):
         #Failure.create(exception)
-        failure = Failure(exception, self._queue, self._payload)
+        failure = Failure(exception, self._queue, self._payload, self._worker)
         failure.save(self.resq)
 
     @classmethod
-    def reserve(cls, queue, res):
+    def reserve(cls, queue, res, worker=None):
         payload = res.pop(queue)
         if payload:
-            return cls(queue, payload, res)
+            return cls(queue, payload, res, worker)

src/pyres/job.py pyres/job.pysrc/pyres/job.py renamed to pyres/job.py

@@ -6,15 +6,16 @@
 import os, sys
 import time
 import simplejson
+
 class Worker(object):
-    def __init__(self, queues=[], server="localhost:6379"):
+    def __init__(self, queues=[], server="localhost:6379", password=None):
         self.queues = queues
         self.validate_queues()
         self._shutdown = False
         self.child = None
         self.pid = os.getpid()
         if isinstance(server,basestring):
-            self.resq = ResQ(server)
+            self.resq = ResQ(server=server, password=password)
         elif isinstance(server, ResQ):
             self.resq = server
         else:
@@ -26,28 +27,29 @@ 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)
         self.started = datetime.datetime.now()
         #Stat.clear("processed:#{self}")
         #Stat.clear("failed:#{self}")
 
     def _set_started(self, time):
         if time:
-            self.resq.redis.set("worker:%s:started" % self, time.strftime('%Y-%m-%d %H:%M:%S'))
+            self.resq.redis.set("resque:worker:%s:started" % self, time.strftime('%Y-%m-%d %H:%M:%S'))
         else:
-            self.resq.redis.delete("worker:%s:started" % self)
-    
+            self.resq.redis.delete("resque:worker:%s:started" % self)
+            
     def _get_started(self):
-        datestring = self.resq.redis.get("worker:%s:started" % self)
+        datestring = self.resq.redis.get("resque:worker:%s:started" % self)
         ds = None
         if datestring:
             ds = datetime.datetime.strptime(datestring, '%Y-%m-%d %H:%M:%S')
         return ds
+    
     started = property(_get_started, _set_started)
 
     def unregister_worker(self):
-        self.resq.redis.srem('workers',str(self))
+        self.resq.redis.srem('resque:workers',str(self))
         self.started = None
 
     def startup(self):
@@ -71,14 +73,13 @@ def kill_child(self, signum, frame):
         if self.child:
             print "Killing child at %s" % self.child
             os.kill(self.child, signal.SIGKILL)
-    
+            
     def __str__(self):
         if getattr(self,'id', None):
             return self.id
         hostname = os.uname()[1]
-        #pid = os.getpid()
         return '%s:%s:%s' % (hostname, self.pid, ','.join(self.queues))
-    
+         
     def work(self, interval=5):
         self.startup()
         while True:
@@ -128,7 +129,7 @@ def process(self, job=None):
     def reserve(self):
         for q in self.queues:
             print "Checking %s" % q
-            job = Job.reserve(q, self.resq)
+            job = Job.reserve(q, self.resq, self.__str__())
             if job:
                 print "Found job on %s" % q
                 return job
@@ -141,14 +142,14 @@ 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
         print "worker:%s" % str(self)
-        print self.resq.redis["worker:%s" % str(self)]
+        print self.resq.redis["resque:worker:%s" % str(self)]
 
     def done_working(self):
         print 'done working'
         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)
@@ -164,20 +165,21 @@ def failed(self):
         stat = Stat("failed:%s" % self, self.resq)
         total_failed.incr()
         stat.incr()
+        
     def get_failed(self):
         return Stat("failed:%s" % self, self.resq).get()
     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 {}
+
 
     def processing(self):
         return self.job()
 
     def state(self):
-        return 'working' if self.resq.redis.exists('worker:%s' % self) else 'idle'
-    
+        return 'working' if self.resq.redis.exists('resque:worker:%s' % self) else 'idle'
 
     @classmethod
     def run(cls, queues, server):
@@ -190,7 +192,7 @@ def all(cls, host="localhost:6379"):
             resq = ResQ(host)
         elif isinstance(host, ResQ):
             resq = host
-        return [Worker.find(w,resq) for w in resq.redis.smembers('workers')]
+        return [Worker.find(w,resq) for w in resq.redis.smembers('resque:workers')]
 
     @classmethod
     def working(cls, host):
@@ -199,13 +201,13 @@ def working(cls, host):
         elif isinstance(host, ResQ):
             resq = host
         total = []
-        for key in Worker.all(host):
-            total.append('worker:%s' % key)
+        for key in Worker.all(host):            
+            total.append('resque:worker:%s' % key)
         names = []
         for key in total:
             value = resq.redis.get(key)
             if value:
-                w = Worker.find(key[7:], resq)
+                w = Worker.find(key[14:], resq) #resque:worker:
                 names.append(w)
         return names
 
@@ -221,7 +223,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__":
@@ -236,4 +238,4 @@ def exists(cls, worker_id, resq):
     queues = options.queue_list.split(',')
     import sys
     sys.path.insert(0,'/Users/mgeorge/dev/pyres/src')
-    Worker.run(queues, options.server)
+    Worker.run(queues, options.server)