Отправить письмо по Успеху/неудачу задачи Сельдерея надежно

Question

В настоящее время я работаю над шаблоном для веб-сервисов в области биоинформатики, используя флакон и сельдерей. Шаблон показывает, как пользователь может создавать задачи и запускать их у работника сельдерея.

Общим требованием к таким службам является информирование пользователя о том, когда задача преуспела или не удалась. Я использую 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!') 

Надеется, что это помогает!