Merge branch 'master' of https://github.com/binarydud/pyres

Author: Whit Morriss

AUTHORS.md

@@ -5,5 +5,6 @@
 * dsc
 * Alex Ezell
 * Michael Russo
+* Whit Morris
 
 Inspired by Resque, by Chris Wanstrath

HISTORY.md

@@ -1,3 +1,13 @@
+## 0.9.1 (2010-10-15)
+* fixing issues #45, #46.
+	* #45 - resweb not working in chrome
+	* #46 - delayed_queue_schedule_size() returns incorrect value
+* updated version requirement for redis-py
+* added Failure docs from Alex._
+
+## 0.9 (2010-08-05)
+* added better logging to the project
+
 ## 0.8 (2010-04-24)
 * added the pyres_manager and the horde module. This allows a more prefork like model for processing jobs.
 * setproctitle usage. Allows better process titles when viewing via ps

docs/source/failures.rst

@@ -0,0 +1,17 @@
+Failures
+===============
+
+Pyres provides a ``BaseBackend`` for handling failed jobs. You can subclass
+this backend to store failed jobs in any system you like.
+
+Currently, the only provided backend is a ``RedisBackend`` which will store
+your failed jobs into a special *failed* queue for later processing or reenqueueing.
+
+Here's a simple example::
+
+	>>> from pyres import failure
+	>>> from pyres.job import Job
+	>>> from pyres import ResQ
+	>>> r = Resq()
+	>>> job = Job.reserve('basic',r)
+	>>> job.fail("problem")

pyres/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '0.9'
+__version__ = '0.9.1'
 
 from redis import Redis
 import pyres.json_parser as json
@@ -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,73 +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):
-        return self.redis.zcard('resque:delayed_queue_schedule')
-    
+        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
@@ -269,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)
@@ -280,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)
@@ -296,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)
-    
+

pyres/failure/__init__.py

@@ -1,32 +1,28 @@
-import datetime
-from pyres import ResQ 
-import sys, traceback
 from pyres.failure.redis import RedisBackend
 
-_backend = RedisBackend
+backend = RedisBackend
 
 def create(*args, **kwargs):
-    return _backend(*args, **kwargs)
+    return backend(*args, **kwargs)
 
 def count(resq):
-    return _backend.count(resq)
+    return backend.count(resq)
 
 def all(resq, start, count):
-    return _backend.all(resq, start, count)
+    return backend.all(resq, start, count)
 
 def clear(resq):
-    return _backend.clear(resq)
+    return backend.clear(resq)
 
 def requeue(resq, failure_object):
     queue = failure_object._queue
     payload = failure_object._payload
     return resq.push(queue, payload)
-    
+
 def retry(resq, queue, payload):
     job = resq.decode(payload)
     resq.push(queue, job['payload'])
     return delete(resq, payload)
-    
+
 def delete(resq, payload):
-    return resq.redis.lrem(name = 'resque:failed', num = 1, value = payload)
-    
+    return resq.redis.lrem(name='resque:failed', num=1, value=payload)