Clean up whitespace.

Author: chawco

pyres/worker.py

@@ -17,10 +17,10 @@ def setproctitle(name):
 class Worker(object):
     """Defines a worker. The ``pyres_worker`` script instantiates this Worker class and
     passes a comma-separated list of queues to listen on.::
-    
+
        >>> from pyres.worker import Worker
        >>> Worker.run([queue1, queue2], server="localhost:6379")
-       
+
     """
     def __init__(self, queues=[], server="localhost:6379", password=None):
         self.queues = queues
@@ -35,40 +35,40 @@ def __init__(self, queues=[], server="localhost:6379", password=None):
             self.resq = server
         else:
             raise Exception("Bad server argument")
-        
-    
+
+
     def validate_queues(self):
         """Checks if a worker is given at least one queue to work on."""
         if not self.queues:
             raise NoQueueError("Please give each worker at least one queue.")
-    
+
     def register_worker(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()
-    
+
     def _set_started(self, dt):
         if dt:
             key = int(time.mktime(dt.timetuple()))
             self.resq.redis.set("resque:worker:%s:started" % self, key)
         else:
             self.resq.redis.delete("resque:worker:%s:started" % self)
-            
+
     def _get_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 datestring
-    
+
     started = property(_get_started, _set_started)
-    
+
     def unregister_worker(self):
         self.resq.redis.srem('resque:workers',str(self))
         self.started = None
         Stat("processed:%s" % self, self.resq).clear()
         Stat("failed:%s" % self, self.resq).clear()
-    
+
     def prune_dead_workers(self):
         all_workers = Worker.all(self.resq)
         known_workers = self.worker_pids()
@@ -85,20 +85,20 @@ def startup(self):
         self.register_signal_handlers()
         self.prune_dead_workers()
         self.register_worker()
-    
+
     def register_signal_handlers(self):
         signal.signal(signal.SIGTERM, self.shutdown_all)
         signal.signal(signal.SIGINT, self.shutdown_all)
         signal.signal(signal.SIGQUIT, self.schedule_shutdown)
         signal.signal(signal.SIGUSR1, self.kill_child)
-    
+
     def shutdown_all(self, signum, frame):
         self.schedule_shutdown(signum, frame)
         self.kill_child(signum, frame)
-    
+
     def schedule_shutdown(self, signum, frame):
         self._shutdown = True
-    
+
     def kill_child(self, signum, frame):
         if self.child:
             logging.info("Killing child at %s" % self.child)
@@ -108,19 +108,19 @@ def __str__(self):
         if getattr(self,'id', None):
             return self.id
         return '%s:%s:%s' % (self.hostname, self.pid, ','.join(self.queues))
-         
+
     def work(self, interval=5):
         """Invoked by ``run`` method. ``work`` listens on a list of queues and sleeps
-        for ``interval`` time. 
-        
+        for ``interval`` time.
+
         ``interval`` -- Number of seconds the worker will wait until processing the next job. Default is "5".
-        
+
         Whenever a worker finds a job on the queue it first calls ``reserve`` on
-        that job to make sure another worker won't run it, then *forks* itself to 
+        that job to make sure another worker won't run it, then *forks* itself to
         work on that job.
-        
+
         Finally, the ``process`` method actually processes the job by eventually calling the Job instance's ``perform`` method.
-        
+
         """
         setproctitle('pyres_worker-%s: Starting' % __version__)
         self.startup()
@@ -157,7 +157,7 @@ def work(self, interval=5):
                 setproctitle("pyres_worker-%s: Waiting for %s " % (__version__, ','.join(self.queues)))
                 time.sleep(interval)
         self.unregister_worker()
-    
+
     def process(self, job=None):
         if not job:
             job = self.reserve()
@@ -174,15 +174,15 @@ def process(self, job=None):
             logging.debug('job details: %s' % job)
         finally:
             self.done_working()
-    
+
     def reserve(self):
         for q in self.queues:
             logging.debug('checking queue %s' % q)
             job = Job.reserve(q, self.resq, self.__str__())
             if job:
                 logging.info('Found job on %s' % q)
                 return job
-    
+
     def working_on(self, job):
         logging.debug('marking as working on')
         data = {
@@ -194,38 +194,38 @@ def working_on(self, job):
         self.resq.redis["resque:worker:%s" % str(self)] = data
         logging.debug("worker:%s" % str(self))
         logging.debug(self.resq.redis["resque:worker:%s" % str(self)])
-    
+
     def done_working(self):
         logging.info('done working')
         self.processed()
         self.resq.redis.delete("resque:worker:%s" % str(self))
-    
+
     def processed(self):
         total_processed = Stat("processed", self.resq)
         worker_processed = Stat("processed:%s" % str(self), self.resq)
         total_processed.incr()
         worker_processed.incr()
-    
+
     def get_processed(self):
         return Stat("processed:%s" % str(self), self.resq).get()
-    
+
     def failed(self):
         Stat("failed", self.resq).incr()
         Stat("failed:%s" % self, self.resq).incr()
-        
+
     def get_failed(self):
         return Stat("failed:%s" % self, self.resq).get()
-    
+
     def job(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('resque:worker:%s' % self) else 'idle'
 
@@ -235,32 +235,32 @@ def worker_pids(self):
         return map(lambda l: l.strip().split(' ')[0],
                    commands.getoutput("ps -A -o pid,command | \
                                        grep pyres_worker").split("\n"))
-    
+
     @classmethod
     def run(cls, queues, server="localhost:6379", interval=None):
         worker = cls(queues=queues, server=server)
         if interval is not None:
-            worker.work(interval)            
+            worker.work(interval)
         else:
             worker.work()
-    
+
     @classmethod
     def all(cls, host="localhost:6379"):
         if isinstance(host,basestring):
             resq = ResQ(host)
         elif isinstance(host, ResQ):
             resq = host
-        
+
         return [Worker.find(w,resq) for w in resq.redis.smembers('resque:workers') or []]
-    
+
     @classmethod
     def working(cls, host):
         if isinstance(host, basestring):
             resq = ResQ(host)
         elif isinstance(host, ResQ):
             resq = host
         total = []
-        for key in Worker.all(host):            
+        for key in Worker.all(host):
             total.append('resque:worker:%s' % key)
         names = []
         for key in total:
@@ -269,7 +269,7 @@ def working(cls, host):
                 w = Worker.find(key[14:], resq) #resque:worker:
                 names.append(w)
         return names
-    
+
     @classmethod
     def find(cls, worker_id, resq):
         if Worker.exists(worker_id, resq):
@@ -279,11 +279,11 @@ def find(cls, worker_id, resq):
             return worker
         else:
             return None
-    
+
     @classmethod
     def exists(cls, worker_id, resq):
         return resq.redis.sismember('resque:workers', worker_id)
-    
+
 
 if __name__ == "__main__":
     from optparse import OptionParser