Merge branch 'feature/failure-backends' into develop

Author: chawco

pyres/failure/__init__.py

@@ -1,32 +1,28 @@
-import datetime
-from pyres import ResQ 
-import sys, traceback
 from pyres.failure.redis import RedisBackend
 
-_backend = RedisBackend
+backend = RedisBackend
 
 def create(*args, **kwargs):
-    return _backend(*args, **kwargs)
+    return backend(*args, **kwargs)
 
 def count(resq):
-    return _backend.count(resq)
+    return backend.count(resq)
 
 def all(resq, start, count):
-    return _backend.all(resq, start, count)
+    return backend.all(resq, start, count)
 
 def clear(resq):
-    return _backend.clear(resq)
+    return backend.clear(resq)
 
 def requeue(resq, failure_object):
     queue = failure_object._queue
     payload = failure_object._payload
     return resq.push(queue, payload)
-    
+
 def retry(resq, queue, payload):
     job = resq.decode(payload)
     resq.push(queue, job['payload'])
     return delete(resq, payload)
-    
+
 def delete(resq, payload):
-    return resq.redis.lrem(name = 'resque:failed', num = 1, value = payload)
-    
+    return resq.redis.lrem(name='resque:failed', num=1, value=payload)

pyres/failure/mail.py

@@ -0,0 +1,103 @@
+import smtplib
+
+from textwrap import dedent
+from email.mime.text import MIMEText
+
+from base import BaseBackend
+
+class MailBackend(BaseBackend):
+    """Extends ``BaseBackend`` to provide support for emailing failures.
+    Intended to be used with the MultipleBackend:
+
+    from pyres import failure
+
+    from pyres.failure.mail import MailBackend
+    from pyres.failure.multiple import MultipleBackend
+    from pyres.failure.redis import RedisBackend
+
+    class EmailFailure(MailBackend):
+        subject = 'Pyres Failure on {queue}'
+        from_user = 'My Email User <[email protected]>'
+        recipients = ['Me <[email protected]>']
+
+        smtp_host = 'mail.mydomain.tld'
+        smtp_port = 25
+        smtp_tls = True
+
+        smtp_user = 'mailuser'
+        smtp_password = 'm41lp455w0rd'
+
+    failure.backend = MultipleBackend
+    failure.backend.classes = [RedisBackend, EmailFailure]
+
+
+    Additional notes:
+        - The following tokens are available in subject: queue, worker, exception
+
+        - Override the create_message method to provide an alternate body. It
+        should return one of the message types from email.mime.*
+    """
+    subject = 'Pyres Failure on {queue}'
+
+    recipients = []
+    from_user = None
+    smtp_host = None
+    smtp_port = 25
+
+    smtp_tls = False
+
+    smtp_user = None
+    smtp_password = None
+
+    def save(self, resq=None):
+        if not self.recipients or not self.smtp_host or not self.from_user:
+            return
+
+        message = self.create_message()
+        subject = self.format_subject()
+
+        message['Subject'] = subject
+        message['From'] = self.from_user
+        message['To'] = ", ".join(self.recipients)
+
+        self.send_message(message)
+
+    def format_subject(self):
+        return self.subject.format(queue=self._queue,
+                                   worker=self._worker,
+                                   exception=self._exception)
+
+    def create_message(self):
+        """Returns a message body to send in this email. Should be from email.mime.*"""
+
+        body = dedent("""\
+        Received exception {exception} on {queue} from worker {worker}:
+
+        {traceback}
+
+        Payload:
+        {payload}
+
+        """).format(exception=self._exception,
+                   traceback=self._traceback,
+                   queue=self._queue,
+                   payload=self._payload,
+                   worker=self._worker)
+
+        return MIMEText(body)
+
+    def send_message(self, message):
+        smtp = smtplib.SMTP(self.smtp_host, self.smtp_port)
+
+        try:
+            smtp.ehlo()
+
+            if self.smtp_tls:
+                smtp.starttls()
+
+            if self.smtp_user:
+                smtp.login(self.smtp_user, self.smtp_password)
+
+            smtp.sendmail(self.from_user, self.recipients, message.as_string())
+        finally:
+            smtp.close()

pyres/failure/multiple.py

@@ -0,0 +1,49 @@
+from base import BaseBackend
+from redis import RedisBackend
+
+class MultipleBackend(BaseBackend):
+    """Extends ``BaseBackend`` to provide support for delegating calls to multiple
+    backends. Queries are delegated to the first backend in the list. Defaults to
+    only the RedisBackend.
+
+    To use:
+
+    from pyres import failure
+
+    from pyres.failure.base import BaseBackend
+    from pyres.failure.multiple import MultipleBackend
+    from pyres.failure.redis import RedisBackend
+
+    class CustomBackend(BaseBackend):
+        def save(self, resq):
+            print('Custom backend')
+
+    failure.backend = MultipleBackend
+    failure.backend.classes = [RedisBackend, CustomBackend]
+    """
+    classes = []
+
+    def __init__(self, *args):
+        if not self.classes:
+            self.classes = [RedisBackend]
+
+        self.backends = [klass(*args) for klass in self.classes]
+        BaseBackend.__init__(self, *args)
+
+    @classmethod
+    def count(cls, resq):
+        first = MultipleBackend.classes[0]
+        return first.count(resq)
+
+    @classmethod
+    def all(cls, resq, start=0, count=1):
+        first = MultipleBackend.classes[0]
+        return first.all(resq, start, count)
+
+    @classmethod
+    def clear(cls, resq):
+        first = MultipleBackend.classes[0]
+        return first.clear(resq)
+
+    def save(self, resq=None):
+        map(lambda x: x.save(resq), self.backends)

pyres/job.py

@@ -1,36 +1,41 @@
 from datetime import datetime, timedelta
 from pyres import ResQ, str_to_class, safe_str_to_class
 from pyres import failure
+from pyres.failure.redis import RedisBackend
 
 class Job(object):
     """Every job on the ResQ is an instance of the *Job* class.
-    
+
     The ``__init__`` takes these keyword arguments:
-    
+
         ``queue`` -- A string defining the queue to which this Job will be added.
-    
+
         ``payload`` -- A dictionary which contains the string name of a class which extends this Job and
         a list of args which will be passed to that class.
-        
+
         ``resq`` -- An instance of the ResQ class.
-    
+
         ``worker`` -- The name of a specific worker if you'd like this Job to be done by that worker. Default is "None".
-    
+
     """
     def __init__(self, queue, payload, resq, worker=None):
         self._queue = queue
         self._payload = payload
         self.resq = resq
         self._worker = worker
-    
+
+        # Set the default back end, jobs can override when we import them
+        # inside perform().
+        failure.backend = RedisBackend
+
     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.
-        
+
         """
         payload_class_str = self._payload["class"]
         payload_class = safe_str_to_class(payload_class_str)
@@ -41,11 +46,11 @@ def perform(self):
         except:
             if not self.retry(payload_class, args):
                 raise
-    
+
     def fail(self, exception):
         """This method provides a way to fail a job and will use whatever failure backend
         you've provided. The default is the ``RedisBackend``.
-        
+
         """
         fail = failure.create(exception, self._queue, self._payload, self._worker)
         fail.save(self.resq)
@@ -70,7 +75,7 @@ def retry(self, payload_class, args):
     def reserve(cls, queue, res, worker=None):
         """Reserve a job on the queue. This marks this job so that other workers
         will not pick it up.
-        
+
         """
         payload = res.pop(queue)
         if payload:

tests/test_failure.py

@@ -1,31 +1,35 @@
 from tests import PyResTests, Basic, TestProcess, ErrorObject
 from pyres import failure
+from pyres.failure.base import BaseBackend
+from pyres.failure.multiple import MultipleBackend
+from pyres.failure.redis import RedisBackend
 from pyres.job import Job
+
 class FailureTests(PyResTests):
     def test_count(self):
         self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
         job.fail("problem")
         assert failure.count(self.resq) == 1
         assert self.redis.llen('resque:failed') == 1
-    
+
     def test_create(self):
         self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
         e = Exception('test')
         fail = failure.create(e, 'basic', job._payload)
         assert isinstance(fail._payload, dict)
-        fail.save()
+        fail.save(self.resq)
         assert failure.count(self.resq) == 1
         assert self.redis.llen('resque:failed') == 1
-    
+
     def test_all(self):
         self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
         e = Exception('problem')
         job.fail(e)
         assert len(failure.all(self.resq, 0, 20)) == 1
-    
+
     def test_clear(self):
         self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
@@ -34,7 +38,7 @@ def test_clear(self):
         assert self.redis.llen('resque:failed') == 1
         failure.clear(self.resq)
         assert self.redis.llen('resque:failed') == 0
-    
+
     def test_requeue(self):
         self.resq.enqueue(Basic,"test1")
         job = Job.reserve('basic',self.resq)
@@ -45,4 +49,21 @@ def test_requeue(self):
         assert self.resq.size('basic') == 1
         job = Job.reserve('basic',self.resq)
         assert job._queue == 'basic'
-        assert job._payload == {'class':'tests.Basic','args':['test1']}
+        assert job._payload == {'class':'tests.Basic','args':['test1']}
+
+# Test the MultipleBackend, basically just repeat the above tests, ensuring that
+# we've delegated to the methods appropriately
+
+class TestBackend(BaseBackend):
+    def save(self, resq):
+        resq.redis.set('testbackend:called', 1)
+
+failure.backend = MultipleBackend
+failure.backend.classes = [RedisBackend, TestBackend]
+
+class MultipleFailureTests(FailureTests):
+    def test_create(self):
+        # Run the parent test
+        FailureTests.test_create(self)
+        # But also ensure the other backends were called
+        assert int(self.redis.get('testbackend:called')) == 1