littlefish/celeryutil.py
def get_enable_celery_error_reporting_function(site_name, from_address):
    """
    Use this to enable error reporting.  You need to put the following in your tasks.py or wherever you
    want to create your celery instance:

    celery = Celery(__name__)

    enable_celery_email_logging = get_enable_celery_error_reporting_function('My Website [LIVE]', 'errors@mywebsite.com')
    after_setup_logger.connect(enable_celery_email_logging)
    after_setup_task_logger.connect(enable_celery_email_logging)
    """
    def enable_celery_email_logging(sender, signal, logger, loglevel, logfile, format, colorize, **kwargs):
        from celery import current_app
        log.info('>> Initialising Celery task error reporting for logger {}'.format(logger.name))
        
        send_errors = current_app.conf['CELERY_SEND_TASK_ERROR_EMAILS']
        send_warnings = current_app.conf['CELERY_SEND_TASK_WARNING_EMAILS']

        if send_errors or send_warnings:
            error_email_subject = '{} Celery ERROR!'.format(site_name)
            celery_handler = CeleryEmailHandler(from_address, current_app.conf['ADMIN_EMAILS'],
                                                error_email_subject)

            if send_warnings:
                celery_handler.setLevel(logging.WARNING)
            else:
                celery_handler.setLevel(logging.ERROR)
            
            logger.addHandler(celery_handler)

    return enable_celery_email_logging
def get_enable_celery_error_reporting_function(site_name, from_address):
    """
    Use this to enable error reporting.  You need to put the following in your tasks.py or wherever you
    want to create your celery instance:

    celery = Celery(__name__)

    enable_celery_email_logging = get_enable_celery_error_reporting_function('My Website [LIVE]', 'errors@mywebsite.com')
    after_setup_logger.connect(enable_celery_email_logging)
    after_setup_task_logger.connect(enable_celery_email_logging)
    """
    def enable_celery_email_logging(sender, signal, logger, loglevel, logfile, format, colorize, **kwargs):
        from celery import current_app
        log.info('>> Initialising Celery task error reporting for logger {}'.format(logger.name))
        
        send_errors = current_app.conf['CELERY_SEND_TASK_ERROR_EMAILS']
        send_warnings = current_app.conf['CELERY_SEND_TASK_WARNING_EMAILS']

        if send_errors or send_warnings:
            error_email_subject = '{} Celery ERROR!'.format(site_name)
            celery_handler = CeleryEmailHandler(from_address, current_app.conf['ADMIN_EMAILS'],
                                                error_email_subject)

            if send_warnings:
                celery_handler.setLevel(logging.WARNING)
            else:
                celery_handler.setLevel(logging.ERROR)
            
            logger.addHandler(celery_handler)

    return enable_celery_email_logging
def get_enable_celery_error_reporting_function ( site_name , from_address ) : def enable_celery_email_logging ( sender , signal , logger , loglevel , logfile , format , colorize , * * kwargs ) : from celery import current_app log . info ( '>> Initialising Celery task error reporting for logger {}' . format ( logger . name ) ) send_errors = current_app . conf [ 'CELERY_SEND_TASK_ERROR_EMAILS' ] send_warnings = current_app . conf [ 'CELERY_SEND_TASK_WARNING_EMAILS' ] if send_errors or send_warnings : error_email_subject = '{} Celery ERROR!' . format ( site_name ) celery_handler = CeleryEmailHandler ( from_address , current_app . conf [ 'ADMIN_EMAILS' ] , error_email_subject ) if send_warnings : celery_handler . setLevel ( logging . WARNING ) else : celery_handler . setLevel ( logging . ERROR ) logger . addHandler ( celery_handler ) return enable_celery_email_logging
Use this to enable error reporting.  You need to put the following in your tasks.py or wherever you
    want to create your celery instance:

    celery = Celery(__name__)

    enable_celery_email_logging = get_enable_celery_error_reporting_function('My Website [LIVE]', 'errors@mywebsite.com')
    after_setup_logger.connect(enable_celery_email_logging)
    after_setup_task_logger.connect(enable_celery_email_logging)
Use this to enable error reporting . You need to put the following in your tasks . py or wherever you want to create your celery instance :