__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!"))