Merge branch 'logging_merge'

Author: Matt George

pyres/__init__.py

@@ -3,6 +3,8 @@
 from redis import Redis
 import pyres.json_parser as json
 
+import logging
+
 def my_import(name):
     """Helper function for walking import calls when searching for classes by string names."""
     mod = __import__(name)    
@@ -73,10 +75,7 @@ class ResQ(object):
     attribute on it.
     
     """
-    def __init__(self, server="localhost:6379", password=None, 
-                 timeout=None, retry_connection=True):
-        self.timeout = timeout
-        #self.retry_connection = retry_connection
+    def __init__(self, server="localhost:6379", password=None):
         self.redis = server
         if password:
             self.redis.auth(password)
@@ -133,14 +132,21 @@ def enqueue(self, klass, *args):
         
         """
         queue = getattr(klass,'queue', None)
-        #print cls._res
         if queue:
             class_name = '%s.%s' % (klass.__module__, klass.__name__)
-            #print class_name
             self.push(queue, {'class':class_name,'args':args})
-            #Job.create(queue, klass,*args)
+            logging.info("enqueued '%s' job" % class_name)
+            if args:
+                logging.debug("job arguments: %s" % args)
+            else:
+                logging.debug("no arguments passed in.")
+        else:
+            logging.warning("unable to enqueue job with class %s" % str(klass))
+
     def enqueue_from_string(self, klass_as_string, queue, *args):
         self.push(queue, {'class':klass_as_string,'args':args})
+        logging.info("enqueued '%s' job" % klass_as_string)
+        logging.debug("job arguments: %s" % args)
 
     def queues(self):
         return self.redis.smembers("resque:queues") or []
@@ -207,13 +213,11 @@ def decode(cls, item):
     @classmethod
     def _enqueue(cls, klass, *args):
         queue = getattr(klass,'queue', None)
-        #print cls._res
         _self = cls()
         if queue:
             class_name = '%s.%s' % (klass.__module__, klass.__name__)
-            #print class_name
             _self.push(queue, {'class':class_name,'args':args})
-            #Job.create(queue, klass,*args)
+
 
 class Stat(object):
     """A Stat class which shows the current status of the queue.

pyres/job.py

@@ -22,6 +22,10 @@ def __init__(self, queue, payload, resq, worker=None):
         self.resq = resq
         self._worker = worker
 
+    def __str__(self):
+        return "(Job{%s} | %s | %s)" % (
+            self._queue, self._payload['class'], repr(self._payload['args']))
+    
     def perform(self):
         """This method converts payload into args and calls the ``perform`` method
         on the payload class.

pyres/worker.py

@@ -1,6 +1,7 @@
 from pyres.exceptions import NoQueueError
 from pyres.job import Job
 from pyres import ResQ, Stat
+import logging
 import signal
 import datetime
 import os, sys
@@ -80,7 +81,7 @@ def schedule_shutdown(self, signum, frame):
 
     def kill_child(self, signum, frame):
         if self.child:
-            print "Killing child at %s" % self.child
+            logging.info("Killing child at %s" % self.child)
             os.kill(self.child, signal.SIGKILL)
 
     def __str__(self):
@@ -105,24 +106,25 @@ def work(self, interval=5):
         self.startup()
         while True:
             if self._shutdown:
-                print 'shutdown scheduled'
+                logging.info('shutdown scheduled')
                 break
             job = self.reserve()
             if job:
-                print "got: %s" % job
+                logging.info('picked up job')
+                logging.debug('job details: %s' % job)
                 self.child = os.fork()
                 if self.child:
-                    print 'Forked %s at %s' % (self.child, datetime.datetime.now())
+                    logging.info('Forked %s at %s' % (self.child, datetime.datetime.now()))
                     try:
                         os.waitpid(self.child, 0)
                     except OSError, ose:
                         import errno
                         if ose.errno != errno.EINTR:
                             raise ose
                     #os.wait()
-                    print 'Done waiting'
+                    logging.debug('done waiting')
                 else:
-                    print 'Processing %s since %s' % (job._queue, datetime.datetime.now())
+                    logging.info('Processing %s since %s' % (job._queue, datetime.datetime.now()))
                     self.process(job)
                     os._exit(0)
                 self.child = None
@@ -140,36 +142,37 @@ def process(self, job=None):
             job.perform()
         except Exception, e:
             exceptionType, exceptionValue, exceptionTraceback = sys.exc_info()
-            print "%s failed: %s" % (job, e)
+            logging.error("%s failed: %s" % (job, e))
             job.fail(exceptionTraceback)
             self.failed()
         else:
-            print "done: %s" % job
+            logging.info('completed job')
+            logging.debug('job details: %s' % job)
         finally:
             self.done_working()
 
     def reserve(self):
         for q in self.queues:
-            print "Checking %s" % q
+            logging.debug('checking queue %s' % q)
             job = Job.reserve(q, self.resq, self.__str__())
             if job:
-                print "Found job on %s" % q
+                logging.info('Found job on %s' % q)
                 return job
 
     def working_on(self, job):
-        print 'marking as working on'
+        logging.debug('marking as working on')
         data = {
             'queue': job._queue,
             'run_at': str(datetime.datetime.now()),
             'payload': job._payload
         }
         data = json.dumps(data)
         self.resq.redis["resque:worker:%s" % str(self)] = data
-        print "worker:%s" % str(self)
-        print self.resq.redis["resque:worker:%s" % str(self)]
+        logging.debug("worker:%s" % str(self))
+        logging.debug(self.resq.redis["resque:worker:%s" % str(self)])
 
     def done_working(self):
-        print 'done working'
+        logging.info('done working')
         self.processed()
         self.resq.redis.delete("resque:worker:%s" % str(self))
 
@@ -203,9 +206,12 @@ def state(self):
         return 'working' if self.resq.redis.exists('resque:worker:%s' % self) else 'idle'
 
     @classmethod
-    def run(cls, queues, server):
+    def run(cls, queues, server, interval):
         worker = cls(queues=queues, server=server)
-        worker.work()
+        if interval is not None:
+            worker.work(interval)            
+        else:
+            worker.work()
 
     @classmethod
     def all(cls, host="localhost:6379"):

scripts/pyres_worker

@@ -1,19 +1,48 @@
 #!/usr/bin/env python
+
+import logging
+
+from optparse import OptionParser
 from pyres.worker import Worker
+
+LOG_LEVELS = {
+    'debug': logging.DEBUG,
+    'info': logging.INFO,
+    'warning': logging.WARNING,
+    'error': logging.ERROR,
+    'critical': logging.CRITICAL
+}
+
 def main():
-    from optparse import OptionParser
+    
     usage = "usage: %prog [options] arg1"
     parser = OptionParser(usage=usage)
     #parser.add_option("-q", dest="queue_list")
     parser.add_option("--host", dest="host", default="localhost")
     parser.add_option("--port",dest="port",type="int", default=6379)
+    parser.add_option("-i", '--interval', dest='interval', default=None, help='the default time interval to sleep between runs')    
+    parser.add_option('-l', '--log-level', dest='log_level', default=LOG_LEVELS['info'], help='log level.  Valid values are "debug", "info", "warning", "error", "critical", in decreasing order of verbosity. Defaults to "info" if parameter not specified.')
     (options,args) = parser.parse_args()
+
     if len(args) != 1:
         parser.print_help()
         parser.error("Argument must be a comma seperated list of queues")
+
+    if options.log_level not in LOG_LEVELS.keys():
+        parser.print_help()
+        parser.error("invalid log level specified")
+    else:
+        log_level = LOG_LEVELS[options.log_level]
+        
+    logging.basicConfig(level=log_level, format="%(asctime)s: %(levelname)s: %(message)s")
+
+    interval = options.interval
+    if interval is not None: 
+        interval = float(interval)
+
     queues = args[0].split(',')
     server = '%s:%s' % (options.host,options.port)
-    Worker.run(queues, server)
+    Worker.run(queues, server, interval)
 
 if __name__ == "__main__":
     main()