Output Dates in Celery Crontab Schedule to Django Template

Question

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.

Answer 1

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)},
    )