saving into extentions branch.

Author: binarymatt

pyres/__init__.py

@@ -96,7 +96,7 @@ def _load_after_fork(mod):
     _after_forks = map(_load_after_fork, _modules)
     return _before_jobs, _after_jobs, _before_forks, _after_forks
 
-_before_jobs, _after_jobs, _before_forks, _after_forks = _load_extensions()
+#_before_jobs, _after_jobs, _before_forks, _after_forks = _load_extensions()
 
 class ResQ(object):
     """The ResQ class defines the Redis server object to which we will

pyres/ext.py

@@ -1,12 +1,15 @@
 import pkgutil
-import pyresext
+import logging
+#import pyresext
 
 def simple_extensions():
+    import pyresext
     extension_modules = [name for _, name, _ in pkgutil.iter_modules([pyresext.__name__])]
     def _load_before_job(mod):
         func = getattr(mod,'before_job', None)
         if func:
             return func
+
     def _load_after_job(mod):
         func = getattr(mod,'after_job', None)
         if func:
@@ -21,7 +24,34 @@ def _load_after_fork(mod):
         func = getattr(mod,'after_fork', None)
         if func:
             return func
-    
+
+    before_job_functions = map(_load_before_job, extension_modules)
+    after_job_functions = map(_load_after_job, extension_modules)
+    before_fork_functions = map(_load_before_fork, extension_modules)
+    after_fork_functions = map(_load_after_fork, extension_modules)
+
+    return before_job_functions, after_job_functions, before_fork_functions, after_fork_functions
+
+before_job_functions, after_job_functions, before_fork_functions, after_fork_functions = simple_extensions()
+
+def _run_job_functions(job, func_list):
+    for func in func_list:
+        try:
+            func(job)
+        except:
+            logging.exception('Error running %s' % func)
+
+def run_before_job(job):
+    _run_job_functions(job, before_job_functions)
+
+def run_after_job(job):
+    _run_job_functions(job, after_job_functions)
+
+def run_before_fork(job):
+    _run_job_functions(job, before_fork_functions)
+
+def run_after_fork(job):
+    _run_job_functions(job, after_fork_functions)
 
 def _load_extensions(package_name='pyresext'):
     """Loads modules from the pyresext."""
@@ -56,5 +86,5 @@ def _load_after_fork(mod):
     _after_forks = map(_load_after_fork, _modules)
     return _before_jobs, _after_jobs, _before_forks, _after_forks
 
-_before_jobs, _after_jobs, _before_forks, _after_forks = _load_extensions()
+#_before_jobs, _after_jobs, _before_forks, _after_forks = _load_extensions()
 

pyres/job.py

@@ -37,8 +37,7 @@ def perform(self):
         on the payload class.
 
         """
-        payload_class_str = self._payload["class"]
-        payload_class = safe_str_to_class(payload_class_str)
+        payload_class = self.job_class()
         payload_class.resq = self.resq
         args = self._payload.get("args", None)
         try:
@@ -71,6 +70,11 @@ def retry(self, payload_class, args):
                 return True
         return False
 
+    def job_class(self):
+        payload_class_str = self._payload["class"]
+        payload_class = safe_str_to_class(payload_class_str)
+        return payload_class
+
     @classmethod
     def reserve(cls, queue, res, worker=None):
         """Reserve a job on the queue. This marks this job so that other workers

pyres/worker.py

@@ -1,6 +1,7 @@
 from pyres.exceptions import NoQueueError
 from pyres.job import Job
 from pyres import ResQ, Stat, __version__
+from pyres.ext import run_before_job, run_after_job, run_before_fork, run_after_fork
 import logging
 import signal
 import datetime, time
@@ -132,6 +133,7 @@ def work(self, interval=5):
             if job:
                 logging.info('picked up job')
                 logging.debug('job details: %s' % job)
+                run_before_fork(job)
                 self.child = os.fork()
                 if self.child:
                     setproctitle("pyres_worker%s: Forked %s at %s" % (__version__, self.child, datetime.datetime.now()))
@@ -145,6 +147,7 @@ def work(self, interval=5):
                     #os.wait()
                     logging.debug('done waiting')
                 else:
+                    run_after_fork(job)
                     setproctitle("pyres_worker-%s: Processing %s since %s" % (__version__, job._queue, datetime.datetime.now()))
                     logging.info('Processing %s since %s' % (job._queue, datetime.datetime.now()))
                     self.process(job)
@@ -163,6 +166,7 @@ def process(self, job=None):
             job = self.reserve()
         try:
             self.working_on(job)
+            run_before_job(job)
             return job.perform()
         except Exception, e:
             exceptionType, exceptionValue, exceptionTraceback = sys.exc_info()
@@ -173,6 +177,7 @@ def process(self, job=None):
             logging.info('completed job')
             logging.debug('job details: %s' % job)
         finally:
+            run_after_job(job)
             self.done_working()
 
     def reserve(self):

pyresext/__init__.py

@@ -0,0 +1,3 @@
+def before_job(job):
+    print 'Before job'
+    return True

pyresext/example.py

@@ -0,0 +1,6 @@
+from tests import PyResTests, Basic, TestProcess, ReturnAllArgsJob
+
+class ExtTests(PyResTests):
+    def test_before_job(self):
+
+        assert False