Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- CELERY_SEND_TASK_ERROR_EMAILS = True
- EMAIL_USE_TLS = True
- EMAIL_HOST_USER = 'noreply@xyz.com'
- EMAIL_HOST_PASSWORD = 'xyz123@'
- DEFAULT_FROM_EMAIL = EMAIL_HOST_USER
- SERVER_EMAIL = EMAIL_HOST_USER
- EMAIL_HOST = 'xyz.abc.com'
- EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
- #!venv/bin/python2
- import os
- from django.conf import settings
- if __name__ == '__main__' and __package__ is None:
- os.sys.path.append(
- os.path.dirname(
- os.path.dirname(
- os.path.abspath(__file__))))
- os.environ.setdefault("DJANGO_SETTINGS_MODULE", "rest_apis.settings")
- import django
- django.setup()
- from django.core.mail import EmailMessage
- from djcelery.models import TaskMeta
- from datetime import datetime, timedelta, time
- USERS_TO_NOTIFY = ['ops@yopmail.com']
- TIME_THRESHOLD_INTERVAL = 60
- def send_email(email_subject_line, email_body):
- email = EmailMessage(email_subject_line,
- email_body,
- settings.EMAIL_HOST_USER,
- USERS_TO_NOTIFY
- )
- email.send()
- def main():
- current_time = datetime.now() # Get Current TimeStamp
- time_threshold = current_time - timedelta(minutes=TIME_THRESHOLD_INTERVAL) # Get 60 minutes past current time stamp
- celery_taskmeta_objects = TaskMeta.objects.filter(status="FAILURE", date_done__gte=time_threshold)
- email_body = "Below are the tasks which failed : "
- if celery_taskmeta_objects.exists():
- for celery_taskmeta in celery_taskmeta_objects:
- print celery_taskmeta.task_id
- email_body += "nntask_id : %s" % celery_taskmeta.task_id
- email_body += "nstatus : %s" % celery_taskmeta.status
- email_body += "ndate : %s" % celery_taskmeta.date_done
- email_body += "ntraceback :"
- email_body += "n%snn" % celery_taskmeta.traceback
- email_subject_line = '[URGENT] Celery task failure in last %s minutes' % (TIME_THRESHOLD_INTERVAL)
- send_email(email_subject_line, email_body)
- main()
- class MyTask(celery.Task):
- def on_failure(self, exc, task_id, args, kwargs, einfo):
- # here you can place mail_admins(...) call
- print('{0!r} failed: {1!r}'.format(task_id, exc))
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement