Merge branch 'develop'

Author: chawco

pyres/__init__.py

@@ -18,19 +18,22 @@ def setup_logging(log_level=logging.INFO, filename=None, stream=sys.stderr):
             handler = WatchedFileHandler(filename)
         except:
             from logging.handlers import RotatingFileHandler
-            handler = RotatingFileHandler(filename,maxBytes=52428800, backupCount=7)
+            handler = RotatingFileHandler(filename,maxBytes=52428800,
+                                          backupCount=7)
     else:
-        handler = logging.StreamHandler(strm=stream)
+        handler = logging.StreamHandler(stream)
     handler.setFormatter(logging.Formatter(
         '%(asctime)s %(levelname)-8s %(message)s', '%Y-%m-%d %H:%M:%S'))
     logger.addHandler(handler)
 
 def my_import(name):
-    """Helper function for walking import calls when searching for classes by string names."""
-    mod = __import__(name)    
-    components = name.split('.')    
-    for comp in components[1:]:        
-        mod = getattr(mod, comp)    
+    """Helper function for walking import calls when searching for classes by
+    string names.
+    """
+    mod = __import__(name)
+    components = name.split('.')
+    for comp in components[1:]:
+        mod = getattr(mod, comp)
     return mod
 
 def safe_str_to_class(s):
@@ -63,34 +66,34 @@ def str_to_class(s):
 class ResQ(object):
     """The ResQ class defines the Redis server object to which we will
     enqueue jobs into various queues.
-    
+
     The ``__init__`` takes these keyword arguments:
-    
+
         ``server`` -- IP address and port of the Redis server to which you want to connect. Default is `localhost:6379`.
-    
+
         ``password`` -- The password, if required, of your Redis server. Default is "None".
-    
+
         ``timeout`` -- The timeout keyword is in the signature, but is unused. Default is "None".
-    
+
         ``retry_connection`` -- This keyword is in the signature but is deprecated. Default is "True".
-    
-    
+
+
     Both ``timeout`` and ``retry_connection`` will be removed as the python-redis client
-    no longer uses them. 
-    
+    no longer uses them.
+
     Example usage::
 
         >>> from pyres import *
         >>> r = ResQ(server="192.168.1.10:6379", password="some_pwd")
             # Assuming redis is running on default port with no password
-    
+
     **r** is a resque object on which we can enqueue tasks.::
 
         >>>> r.enqueue(SomeClass, args)
 
-    SomeClass can be any python class with a *perform* method and a *queue* 
+    SomeClass can be any python class with a *perform* method and a *queue*
     attribute on it.
-    
+
     """
     def __init__(self, server="localhost:6379", password=None):
         self.redis = server
@@ -144,9 +147,9 @@ def _set_redis(self, server):
     redis = property(_get_redis, _set_redis)
 
     def enqueue(self, klass, *args):
-        """Enqueue a job into a specific queue. Make sure the class you are passing
-        has **queue** attribute and a **perform** method on it.
-        
+        """Enqueue a job into a specific queue. Make sure the class you are
+        passing has **queue** attribute and a **perform** method on it.
+
         """
         queue = getattr(klass,'queue', None)
         if queue:
@@ -170,14 +173,14 @@ def enqueue_from_string(self, klass_as_string, queue, *args, **kwargs):
             logging.debug("job arguments: %s" % str(args))
         else:
             logging.debug("no arguments passed in.")
-    
+
     def queues(self):
         return self.redis.smembers("resque:queues") or []
-    
+
     def info(self):
-        """Returns a dictionary of the current status of the pending jobs, 
+        """Returns a dictionary of the current status of the pending jobs,
         processed, no. of queues, no. of workers, no. of failed jobs.
-        
+
         """
         pending = 0
         for q in self.queues():
@@ -191,77 +194,81 @@ def info(self):
             'failed'    : Stat('failed',self).get(),
             'servers'   : ['%s:%s' % (self.redis.host, self.redis.port)]
         }
-    
+
     def keys(self):
-        return [key.replace('resque:','') for key in self.redis.keys('resque:*')]
-    
+        return [key.replace('resque:','')
+                for key in self.redis.keys('resque:*')]
+
     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):
         from pyres.worker import Worker
         return Worker.all(self)
-    
+
     def working(self):
         from pyres.worker import Worker
         return Worker.working(self)
-    
+
     def remove_queue(self, queue):
         if queue in self._watched_queues:
             self._watched_queues.remove(queue)
         self.redis.srem('resque:queues',queue)
         del self.redis['resque:queue:%s' % queue]
-    
+
     def close(self):
         """Close the underlying redis connection.
-        
+
         """
         self.redis.disconnect()
-    
+
     def enqueue_at(self, datetime, klass, *args, **kwargs):
         class_name = '%s.%s' % (klass.__module__, klass.__name__)
-        logging.info("enqueued '%s' job for execution at %s" % (class_name, datetime))
+        logging.info("enqueued '%s' job for execution at %s" % (class_name,
+                                                                datetime))
         if args:
             logging.debug("job arguments are: %s" % str(args))
         payload = {'class':class_name, 'queue': klass.queue, 'args':args}
         if 'first_attempt' in kwargs:
             payload['first_attempt'] = kwargs['first_attempt']
         self.delayed_push(datetime, payload)
-    
+
     def delayed_push(self, datetime, item):
         key = int(time.mktime(datetime.timetuple()))
         self.redis.rpush('resque:delayed:%s' % key, ResQ.encode(item))
         self.redis.zadd('resque:delayed_queue_schedule', key, key)
-    
+
     def delayed_queue_peek(self, start, count):
-        return [int(item) for item in self.redis.zrange('resque:delayed_queue_schedule', start, start+count) or []]
-    
+        return [int(item) for item in self.redis.zrange(
+                'resque:delayed_queue_schedule', start, start+count) or []]
+
     def delayed_timestamp_peek(self, timestamp, start, count):
         return self.list_range('resque:delayed:%s' % timestamp, start, count)
-        
+
     def delayed_queue_schedule_size(self):
         size = 0
         length = self.redis.zcard('resque:delayed_queue_schedule')
         for i in self.redis.zrange('resque:delayed_queue_schedule',0,length):
             size += self.delayed_timestamp_size(i)
         return size
-        
+
     def delayed_timestamp_size(self, timestamp):
         #key = int(time.mktime(timestamp.timetuple()))
         return self.redis.llen("resque:delayed:%s" % timestamp)
-    
+
     def next_delayed_timestamp(self):
         key = int(time.mktime(ResQ._current_time().timetuple()))
-        array = self.redis.zrangebyscore('resque:delayed_queue_schedule', '-inf', key)
+        array = self.redis.zrangebyscore('resque:delayed_queue_schedule',
+                                         '-inf', key)
         timestamp = None
         if array:
             timestamp = array[0]
         return timestamp
-    
+
     def next_item_for_timestamp(self, timestamp):
         #key = int(time.mktime(timestamp.timetuple()))
         key = "resque:delayed:%s" % timestamp
@@ -273,7 +280,7 @@ def next_item_for_timestamp(self, timestamp):
             self.redis.delete(key)
             self.redis.zrem('resque:delayed_queue_schedule', timestamp)
         return item
-    
+
     @classmethod
     def encode(cls, item):
         return json.dumps(item)
@@ -284,7 +291,7 @@ def decode(cls, item):
             ret = json.loads(item)
             return ret
         return None
-    
+
     @classmethod
     def _enqueue(cls, klass, *args):
         queue = getattr(klass,'queue', None)
@@ -300,25 +307,25 @@ def _current_time():
 
 class Stat(object):
     """A Stat class which shows the current status of the queue.
-    
+
     """
     def __init__(self, name, resq):
         self.name = name
         self.key = "resque:stat:%s" % self.name
         self.resq = resq
-    
+
     def get(self):
         val = self.resq.redis.get(self.key)
         if val:
             return int(val)
         return 0
-    
+
     def incr(self, ammount=1):
         self.resq.redis.incr(self.key, ammount)
-    
+
     def decr(self, ammount=1):
         self.resq.redis.decr(self.key, ammount)
-    
+
     def clear(self):
         self.resq.redis.delete(self.key)
-    
+