Merge commit 'ee677c0236f116847f7e6852cedb9d21c6b90f76' into refactor_merge

Conflicts:
	HISTORY.md
	pyres/__init__.py

Author: Matt George

pyres/__init__.py

@@ -4,6 +4,28 @@
 import simplejson
 
 import types
+
+def my_import(name):
+    mod = __import__(name)    
+    components = name.split('.')    
+    for comp in components[1:]:        
+        mod = getattr(mod, comp)    
+    return mod
+
+def safe_str_to_class(s):
+    lst = s.split(".")
+    klass = lst[-1]
+    mod_list = lst[:-1]
+    module = ".".join(mod_list)
+    try:
+        mod = my_import(module)
+        if hasattr(mod, klass):
+            return getattr(mod, klass)
+        else:
+            return None
+    except ImportError:
+        return None
+
 def str_to_class(s):
     lst = s.split(".")
     klass = lst[-1]
@@ -20,7 +42,10 @@ def str_to_class(s):
 
 class ResQ(object):
 
-    def __init__(self, server="localhost:6379", password=None):
+    def __init__(self, server="localhost:6379", password=None, 
+                 timeout=None, retry_connection=True):
+        self.timeout = timeout
+        self.retry_connection = retry_connection
         self.redis = server
         if password:
             self.redis.auth(password)
@@ -63,7 +88,9 @@ def _set_redis(self, server):
         if isinstance(server, basestring):
             self.dsn = server
             host, port = server.split(':')
-            self._redis = Redis(host=host, port=int(port))
+            self._redis = Redis(host=host, port=int(port), 
+                                retry_connection=self.retry_connection,
+                                timeout=self.timeout)
         elif isinstance(server, Redis):
             self.dsn = '%s:%s' % (server.host,server.port)
             self._redis = server

pyres/failure.py

@@ -2,12 +2,12 @@
 from pyres import ResQ 
 import sys, traceback
 class Failure(object):
-    def __init__(self, exp, queue, payload):
+    def __init__(self, exp, worker, queue, payload):
         excc, _, tb = sys.exc_info()
 
         self._exception = excc
         self._traceback = tb
-        #self._worker = worker
+        self._worker = worker
         self._queue = queue
         self._payload = payload
 
@@ -30,7 +30,8 @@ def save(self, resq):
             'payload'   : self._payload,
             'error'     : self._parse_message(self._exception),
             'backtrace' : self._parse_traceback(self._traceback),
-            'queue'     : self._queue
+            'queue'     : self._queue,
+            'worker'    : self._worker
         }
         data = ResQ.encode(data)
         resq.redis.push('resque:failed', data)

pyres/job.py

@@ -1,14 +1,15 @@
-from pyres import ResQ, str_to_class
+from pyres import ResQ, str_to_class, safe_str_to_class
 from pyres.failure import Failure
 class Job(object):
-    def __init__(self, queue, payload, resq):
+    def __init__(self, queue, payload, resq, worker):
         self._queue = queue
         self._payload = payload
         self.resq = resq
+        self._worker = worker
 
     def perform(self):
         payload_class_str = self._payload["class"]
-        payload_class = str_to_class(payload_class_str)
+        payload_class = safe_str_to_class(payload_class_str)
         args = self._payload.get("args", None)
         if args:
             return payload_class.perform(*args)
@@ -17,11 +18,11 @@ def perform(self):
 
     def fail(self, exception):
         #Failure.create(exception)
-        failure = Failure(exception, self._queue, self._payload)
+        failure = Failure(exception, self._worker, self._queue, self._payload)
         failure.save(self.resq)
 
     @classmethod
-    def reserve(cls, queue, res):
+    def reserve(cls, queue, res, worker):
         payload = res.pop(queue)
         if payload:
-            return cls(queue, payload, res)
+            return cls(queue, payload, res, worker)

pyres/worker.py

@@ -129,7 +129,7 @@ def process(self, job=None):
     def reserve(self):
         for q in self.queues:
             print "Checking %s" % q
-            job = Job.reserve(q, self.resq)
+            job = Job.reserve(q, self.resq, self.__str__())
             if job:
                 print "Found job on %s" % q
                 return job