I’m using Celery for my Django project and I’ve scheduled some crontab tasks to send emails out to users at certain times. I need to output a schedule in an HTML/Django template that shows the dates that users can expect the emails to go out on. My crontab schedule looks like this for Celery:
import os from celery import Celery from celery.schedules import crontab app = Celery("myapp") app.config_from_object("django.conf:settings", namespace="CELERY") app.autodiscover_tasks() @app.task(bind=True) def debug_task(self): print("Request: {0!r}".format(self.request)) first_monday_april_to_july = crontab( minute=15, hour=6, day_of_week="monday", day_of_month="1-7", month_of_year="4,5,6,7" ) august_every_monday_first_two_weeks = crontab( minute=15, hour=6, day_of_week="monday", day_of_month="1-14", month_of_year="8" ) august_weekdays_second_two_weeks = crontab( minute=15, hour=6, day_of_week="mon-fri", day_of_month="15-31", month_of_year="8" ) app.conf.beat_schedule = { "report1": { "task": "email_report", "schedule": first_monday_april_to_july, }, "report2": { "task": "email_report", "schedule": august_every_monday_first_two_weeks, }, "report3": { "task": "email_report", "schedule": august_weekdays_second_two_weeks, }
I was hoping to be able to grab all of the dates that the emails will be sent on from these tasks and save them into a list in the view and then send it to the template to be displayed in a table. Is something like this possible? I’m not having any luck finding a way to do this so far.
Advertisement
Answer
I ended up creating a function to grab all of the dates from the tasks and then append them all to a schedule list and then passed that through the view to the template to be output in a table. It looks like this:
from datetime import date, datetime, timedelta from .celery import ( first_monday_april_to_july, august_every_monday_first_two_weeks, august_weekdays_second_two_weeks, ) def get_schedule_due_dates(schedule, last_date=datetime.now().date().replace(month=12, day=31)): start_date = datetime.now() end_date = datetime.combine(last_date, datetime.max.time()) schedules = [] while start_date <= end_date: next_due_schedule = schedule.remaining_estimate(start_date) due_date = datetime.now() + next_due_schedule if due_date <= end_date: schedules.append(due_date) else: break start_date = due_date + timedelta(days=1) return schedules def report_schedule(request): schedules = [] schedules.extend(get_schedule_due_dates(first_monday_april_to_july)) schedules.extend(get_schedule_due_dates(august_every_monday_first_two_weeks)) schedules.extend(get_schedule_due_dates(august_weekdays_second_two_weeks)) return render( request, "reports/report_schedule.html", {"schedules": sorted(schedules)}, )