add enqueue & perform timestamps, add job hooks for accessing them

Classes which implement the perform() method can now also implement
before_perform() and after_perform() methods which take a metadata dict.
These contain the args passed in, as well as timestamps for when the job
was enqueued and when it was performed, and whether the job failed and
was retried.

Author: joeshaw

pyres/__init__.py

@@ -212,7 +212,8 @@ def enqueue(self, klass, *args):
         queue = getattr(klass,'queue', None)
         if queue:
             class_name = '%s.%s' % (klass.__module__, klass.__name__)
-            self.push(queue, {'class':class_name,'args':args})
+            self.push(queue, {'class':class_name,'args':args,
+                              'enqueue_timestamp':time.time()})
             logging.info("enqueued '%s' job on queue %s" % (class_name, queue))
             if args:
                 logging.debug("job arguments: %s" % str(args))
@@ -222,7 +223,8 @@ def enqueue(self, klass, *args):
             logging.warning("unable to enqueue job with class %s" % str(klass))
 
     def enqueue_from_string(self, klass_as_string, queue, *args, **kwargs):
-        payload = {'class':klass_as_string, 'queue': queue, 'args':args}
+        payload = {'class':klass_as_string, 'queue': queue, 'args':args,
+                   'enqueue_timestamp':time.time()}
         if 'first_attempt' in kwargs:
             payload['first_attempt'] = kwargs['first_attempt']
         self.push(queue, payload)
@@ -356,7 +358,8 @@ def _enqueue(cls, klass, *args):
         _self = cls()
         if queue:
             class_name = '%s.%s' % (klass.__module__, klass.__name__)
-            _self.push(queue, {'class':class_name,'args':args})
+            _self.push(queue, {'class':class_name,'args':args,
+                               'enqueue_timestamp': time.time()})
 
     @staticmethod
     def _current_time():

pyres/job.py

@@ -1,3 +1,4 @@
+import time
 from datetime import timedelta
 from pyres import ResQ, safe_str_to_class
 from pyres import failure
@@ -31,6 +32,8 @@ def __init__(self, queue, payload, resq, worker=None):
         self.resq = resq
         self._worker = worker
 
+        self.enqueue_timestamp = self._payload.get("enqueue_timestamp")
+
         # Set the default back end, jobs can override when we import them
         # inside perform().
         failure.backend = RedisBackend
@@ -43,6 +46,20 @@ def perform(self):
         """This method converts payload into args and calls the ``perform``
         method on the payload class.
 
+        Before calling ``perform``, a ``before_perform`` class method
+        is called, if it exists.  It takes a dictionary as an argument;
+        currently the only things stored on the dictionary are the
+        args passed into ``perform`` and a timestamp of when the job
+        was enqueued.
+
+        Similarly, an ``after_perform`` class method is called after
+        ``perform`` is finished.  The metadata dictionary contains the
+        same data, plus a timestamp of when the job was performed, a
+        ``failed`` boolean value, and if it did fail, a ``retried``
+        boolean value.  This method is called after retry, and is
+        called regardless of whether an exception is ultimately thrown
+        by the perform method.
+
         #@ add entry_point loading
 
         """
@@ -51,11 +68,29 @@ def perform(self):
         payload_class.resq = self.resq
         args = self._payload.get("args")
 
+        metadata = dict(args=args)
+        if self.enqueue_timestamp:
+            metadata["enqueue_timestamp"] = self.enqueue_timestamp
+
+        before_perform = getattr(payload_class, "before_perform", None)
+        if before_perform:
+            before_perform(metadata)
+
+        metadata["failed"] = False
+        metadata["perform_timestamp"] = time.time()
         try:
             return payload_class.perform(*args)
         except:
+            metadata["failed"] = True
             if not self.retry(payload_class, args):
+                metadata["retried"] = False
                 raise
+            else:
+                metadata["retried"] = True
+        finally:
+            after_perform = getattr(payload_class, "after_perform", None)
+            if after_perform:
+                after_perform(metadata)
 
     def fail(self, exception):
         """This method provides a way to fail a job and will use whatever