可靠地发送邮件在芹菜任务成功/失败

Question

我目前正在使用Flask和芹菜生物信息网络服务的模板。该模板显示用户如何创建任务并在芹菜工作人员上运行这些任务。

这种服务的一个常见要求是在任务成功或失败时通知用户。我正在使用Flask-Mail发送邮件。我的第一次尝试是这样的：

@celery.task(name='app.expensive_greet', bind=True) 
def expensive_greet(self, person, total, email): 
    this_task_id = expensive_greet.request.id 

    try: 
     for i in range(total): 
      time.sleep(0.1) 
      self.update_state(state='PROGRESS', 
           meta={'current': i, 'total': total}) 

     if email: 
      send_success_mail(email, this_task_id) 
     return 'Greetings, {}!'.format(person) 
    except SoftTimeLimitExceeded: 
     if email: 
      send_failure_mail(email, this_task_id) 
     return 'Greetings, fallback!' 
    except Exception: 
     if email: 
      send_failure_mail(email, this_task_id) 

正如你看到的有很多重复的代码。我想知道是否有可能在邮件处理在自定义芹菜Task隔离并结束了与此：

class MailBase(celery.Task): 
    abstract = True 
    def on_success(self, res, task_id, args, kwargs): 
     _, _, email = args 
     if email: 
      send_success_mail(email, task_id) 

    def on_failure(self, exc, task_id, args, kwargs, einf): 
     _, _, email = args 
     if email: 
      send_failure_mail(email, task_id) 

和设置base=MailBase的任务。我不喜欢这个解决方案，因为：

1. 它仍然有重复的if，但我可以忍受。
2. 如果用户更改任务函数的参数，则还必须在两个不同的位置修改MailBase类。
3. 解开函数参数是很难看的。
4. 由于这是大多数非技术用户的模板，因此简单性很重要。

有没有更好的方法来做到这一点？理想情况下，我希望将邮件作为元数据与任务一起发送，而不是任务函数的参数，并且能够在不触及任务本身的情况下插入邮件功能。

在此先感谢！

Answer 1

如果您只想在任务失败时得到通知，则可以使用芹菜built-in email mechanism。

另外，如果你仍然想坚持自己的方式，你可以尝试使用装饰器来包装电子邮件相关的操作。

import functools 


def send_emails(func): 
    @functools.wraps(func) 
    def wrapper(self, *args, **kwargs): 
     this_task_id = self.request.id 
     email = kwargs.pop('email', False) # get email and remove it from kwargs 
     try: 
      ret = func(self, *args, **kwargs) 
     except NotifyException as ex: 
      if email: 
       send_failure_mail(email, this_task_id) 
      return ex.value 
     except Exception: 
      if email: 
       send_failure_mail(email, this_task_id) 
      # It would be better to raise again to allow celery knows the task has failed 
      raise 
     else: 
      if email: 
       send_success_mail(mail, this_task_id) 
      return ret 
    return wrapper 

NotifyException介绍了一种情况下，当出现错误，但用户并没有把它当作一个失败，只是想发送一封电子邮件来代替。

class NotifyException(Exception): 
    ''' 
    This exception would be handled by send_emails decorator, the decorator will 
    catch it and return its value to outer. 
    ''' 
    def __init__(self, value): 
     self.value = value 
     super(NotifyException, self).__init__(value) 

请注意，最好在配置文件中保存email参数。

而现在的任务方法是这样

@celery.task(name='app.expensive_greet', bind=True) 
@send_emails 
def expensive_greet(self, person, total): 
    try: 
     for i in range(total): 
      time.sleep(0.1) 
      self.update_state(state='PROGRESS', 
           meta={'current': i, 'total': total}) 
     return 'Greetings, {}!'.format(person) 
    except SoftTimeLimitExceeded: 
     # Note that raise this exception to allow the decorator catch it 
     # and return the value of the exception 
     raise NotifyException('Greetings, fallback!') 

改变希望它能帮助！