Merge branch 'chase-seibert/error-email-send-task-sender-class'

Conflicts:
	celery/app/task/__init__.py
	celery/worker/job.py

celery/app/task/__init__.py

@@ -10,6 +10,7 @@ from ...execute.trace import TaskTrace
 from ...registry import tasks, _unpickle_task
 from ...result import EagerResult
 from ...utils import fun_takes_kwargs, mattrgetter, uuid
+from ...utils.mail import ErrorMailSender
 
 extract_exec_options = mattrgetter("queue", "routing_key",
                                    "exchange", "immediate",
@@ -17,7 +18,6 @@ extract_exec_options = mattrgetter("queue", "routing_key",
                                    "serializer", "delivery_mode",
                                    "compression")
 
-
 class Context(threading.local):
     # Default context
     logfile = None
@@ -180,6 +180,7 @@ class BaseTask(object):
 
     #: List of exception types to send error emails for.
     error_whitelist = ()
+    ErrorMailSenderClass = ErrorMailSender
 
     #: The name of a serializer that are registered with
     #: :mod:`kombu.serialization.registry`.  Default is `"pickle"`.
@@ -661,6 +662,11 @@ class BaseTask(object):
         """
         pass
 
+    def send_error_email(self, context, exc, **kwargs):
+        if self.send_error_emails and not self.disable_error_emails:
+            sender = self.ErrorMailSenderClass(self, **kwargs)
+            sender.send(context, exc)
+
     def on_success(self, retval, task_id, args, kwargs):
         """Success handler.
 

celery/tests/test_app/test_app.py

@@ -15,6 +15,8 @@ from celery.utils.serialization import pickle
 from celery.tests import config
 from celery.tests.utils import (unittest, mask_modules, platform_pyimp,
                                 sys_platform, pypy_version)
+from celery.utils.mail import ErrorMailSender
+from kombu.utils import gen_unique_id
 
 THIS_IS_A_KEY = "this is a value"
 
@@ -242,6 +244,13 @@ class test_App(unittest.TestCase):
                                        routing_key="bar_exchange"))
         self.assertIn("bar_exchange", amqp._exchanges_declared)
 
+    def test_error_mail_sender(self):
+        x = ErrorMailSender.subject % {"name": "task_name",
+                                       "id": gen_unique_id(),
+                                       "exc": "FOOBARBAZ",
+                                       "hostname": "lana"}
+        self.assertTrue(x)
+
 
 class test_BaseApp(unittest.TestCase):
 

celery/tests/test_worker/test_worker_job.py

@@ -513,11 +513,6 @@ class test_TaskRequest(unittest.TestCase):
                            "exc": "FOOBARBAZ",
                            "traceback": "foobarbaz"}
         self.assertTrue(x)
-        x = tw.email_subject % {"name": tw.task_name,
-                                     "id": tw.task_id,
-                                     "exc": "FOOBARBAZ",
-                                     "hostname": "lana"}
-        self.assertTrue(x)
 
     def test_from_message(self):
         body = {"task": mytask.name, "id": uuid(),

celery/utils/mail.py

@@ -1,5 +1,6 @@
 import sys
 import smtplib
+from celery.utils import get_symbol_by_name
 
 try:
     from email.mime.text import MIMEText
@@ -77,3 +78,55 @@ class Mailer(object):
 
         client.sendmail(message.sender, message.to, str(message))
         client.quit()
+
+class ErrorMailSender(object):
+
+    # pep8.py borks on a inline signature separator and
+    # says "trailing whitespace" ;)
+    EMAIL_SIGNATURE_SEP = "-- "
+
+    #: Format string used to generate error email subjects.
+    subject = """\
+        [celery@%(hostname)s] Error: Task %(name)s (%(id)s): %(exc)s
+    """
+
+    #: Format string used to generate error email content.
+    body = """
+Task %%(name)s with id %%(id)s raised exception:\n%%(exc)r
+
+
+Task was called with args: %%(args)s kwargs: %%(kwargs)s.
+
+The contents of the full traceback was:
+
+%%(traceback)s
+
+%(EMAIL_SIGNATURE_SEP)s
+Just to let you know,
+celeryd at %%(hostname)s.
+""" % {"EMAIL_SIGNATURE_SEP": EMAIL_SIGNATURE_SEP}
+
+    error_whitelist = None
+
+    def __init__(self, task, **kwargs):
+        #subject=None, body=None, error_whitelist=None
+        self.task = task
+        self.email_subject = kwargs.get("subject", self.subject)
+        self.email_body = kwargs.get("body", self.body)
+        self.error_whitelist = getattr(task, "error_whitelist")
+
+    def should_send(self, context, exc):
+        allow_classes = tuple(map(get_symbol_by_name,  self.error_whitelist))
+        return not self.error_whitelist or isinstance(exc, allow_classes)
+
+    def format_subject(self, context):
+        return self.subject.strip() % context
+
+    def format_body(self, context):
+        return self.body.strip() % context
+
+    def send(self, context, exc, fail_silently=True):
+        if self.should_send(context, exc):
+            self.task.app.mail_admins(self.format_subject(context),
+                                      self.format_body(context),
+                                      fail_silently=fail_silently)

celery/worker/job.py

@@ -15,34 +15,12 @@ from .. import registry
 from ..app import app_or_default
 from ..datastructures import ExceptionInfo
 from ..execute.trace import TaskTrace
-from ..utils import (noop, kwdict, fun_takes_kwargs,
-                     get_symbol_by_name, truncate_text)
+from ..utils import noop, kwdict, fun_takes_kwargs, truncate_text
 from ..utils.encoding import safe_repr, safe_str, default_encoding
 from ..utils.timeutils import maybe_iso8601
 
 from . import state
 
-# pep8.py borks on a inline signature separator and
-# says "trailing whitespace" ;)
-EMAIL_SIGNATURE_SEP = "-- "
-
-#: format string for the body of an error email.
-TASK_ERROR_EMAIL_BODY = """
-Task %%(name)s with id %%(id)s raised exception:\n%%(exc)r
-
-
-Task was called with args: %%(args)s kwargs: %%(kwargs)s.
-
-The contents of the full traceback was:
-
-%%(traceback)s
-
-%(EMAIL_SIGNATURE_SEP)s
-Just to let you know,
-celeryd at %%(hostname)s.
-""" % {"EMAIL_SIGNATURE_SEP": EMAIL_SIGNATURE_SEP}
-
-
 #: Keys to keep from the message delivery info.  The values
 #: of these keys must be pickleable.
 WANTED_DELIVERY_INFO = ("exchange", "routing_key", "consumer_tag", )
@@ -232,13 +210,8 @@ class TaskRequest(object):
     #: Format string used to log task retry.
     retry_msg = """Task %(name)s[%(id)s] retry: %(exc)s"""
 
-    #: Format string used to generate error email subjects.
-    email_subject = """\
-        [celery@%(hostname)s] Error: Task %(name)s (%(id)s): %(exc)s
-    """
-
-    #: Format string used to generate error email content.
-    email_body = TASK_ERROR_EMAIL_BODY
+    email_subject = None
+    email_body = None
 
     #: Timestamp set when the task is started.
     time_start = None
@@ -269,8 +242,8 @@ class TaskRequest(object):
         self.hostname = hostname or socket.gethostname()
         self.logger = logger or self.app.log.get_default_logger()
         self.eventer = eventer
-        self.email_subject = email_subject or self.email_subject
-        self.email_body = email_body or self.email_body
+        self.email_subject = email_subject
+        self.email_body = email_body
 
         self.task = registry.tasks[self.task_name]
         self._store_errors = True
@@ -516,9 +489,10 @@ class TaskRequest(object):
                                           "hostname": self.hostname}})
 
         task_obj = registry.tasks.get(self.task_name, object)
-        self.send_error_email(task_obj, context, exc_info.exception,
-                              enabled=task_obj.send_error_emails,
-                              whitelist=task_obj.error_whitelist)
+        task_obj.send_error_email(context,
+                                  exc_info.exception,
+                                  subject=self.email_subject,
+                                  body=self.email_body)
 
     def acknowledge(self):
         """Acknowledge task."""
@@ -526,16 +500,6 @@ class TaskRequest(object):
             self.on_ack()
             self.acknowledged = True
 
-    def send_error_email(self, task, context, exc,
-            whitelist=None, enabled=False, fail_silently=True):
-        if enabled and not task.disable_error_emails:
-            if not whitelist or isinstance(exc,
-                    tuple(map(get_symbol_by_name, whitelist))):
-                subject = self.email_subject.strip() % context
-                body = self.email_body.strip() % context
-                self.app.mail_admins(subject, body,
-                                     fail_silently=fail_silently)
-
     def repr_result(self, result, maxlen=46):
         # 46 is the length needed to fit
         #     "the quick brown fox jumps over the lazy dog" :)