celery.py 文件源码

def shared_task_email(func):
    """
    Replacement for @shared_task decorator that emails admins if an exception is raised.
    """
    @wraps(func)
    def new_func(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except:
            subject = "Celery task failure"
            message = traceback.format_exc()
            mail_admins(subject, message)
            raise
    return shared_task(new_func)