Source code for scipost.management.commands.check_celery

__copyright__ = "Copyright © Stichting SciPost (SciPost Foundation)"
__license__ = "AGPL v3"


import datetime

from django.core import mail
from django.core.management.base import BaseCommand
from django.utils import timezone

from django_celery_results.models import TaskResult


[docs]class Command(BaseCommand): help = "Check if Celery is still running, or at least not failing."
[docs] def handle(self, *args, **kwargs): # check failed. compare_dt = timezone.now() - datetime.timedelta(hours=1) results_failed = ( TaskResult.objects.filter(status="FAILURE", date_done__gt=compare_dt) .order_by("date_done") .last() ) if results_failed: # Mail failed body = "Celery has failed task results. Last failed ID: {}".format( results_failed.id ) mail.mail_admins("Celery failed", body) self.stdout.write( self.style.SUCCESS( "Celery failed, last ID: {}.".format(results_failed.id) ) ) else: last_result = ( TaskResult.objects.filter(date_done__gt=compare_dt) .order_by("date_done") .last() ) if last_result and last_result.date_done < compare_dt: # Mail inactive body = "No results for Celery found. Celery seems to be inactive." body += " Last result ID: {}".format(last_result.id) mail.mail_admins("Celery inactive", body) self.stdout.write( self.style.SUCCESS( "Celery inactive, last ID: {}.".format(last_result.id) ) ) if not results_failed and not last_result: self.stdout.write(self.style.SUCCESS("Celery alive!"))