Sending Scheduled Reports Via Email Using Django And Celery

30 May 2017

A Guide to Sending Scheduled Reports Via Email Using Django And Celery

In this blog post, we will implement sending out scheduled reports via email to our customers in a Django application using Django and celery.

What is Celery?

We will be using Celery to schedule our reports. Celery is an asynchronous task queue based on distributed message passing. It also supports scheduling of tasks. There are two parts in Celery:

Worker – Entity which manages the running of tasks in Celery.
Broker – Celery communicates through messages, it is the job if the broker to mediate messages between client and worker. Some of the brokers are RabbitMQ and Redis.

Installing Celery

For Django projects, we will install django-celery which in turn installs celery as a dependency. Run this command to install Django-celery: pip install django-celery

Configuring Celery

Adding Django celery configuration in settings.py:

import djcelery
//Add django celery in install apps
INSTALLED_APPS = [
    ...
    'djcelery',
    //Django based broker
    'kombu.transport.django',
    ...
]
# Celery settings
//Specify which broker you will use, we are using django's broker for development
BROKER_URL = 'django://'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend'
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"
djcelery.setup_loader()

Assuming this is your project structure, we will create a Celery instance in the project folder called celeryapp.py.

project
└─── project/__init__.py
└─── project/settings.py
└─── project/urls.py
└ manage.py

Define your celery instance in project/project/celeryapp.py

from __future__ import absolute_import
import os
from celery import Celery
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'project.settings')
from django.conf import settings
app = Celery('project')
# Using a string here means the worker don't have to serialize
# the configuration object to child processes.
app.config_from_object('django.conf:settings')
# Load task modules from all registered Django app configs.
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

Now we need to make sure that celery is loaded when your Django application starts. To ensure this import your celery instance in project/project/init.py

from __future__ import absolute_import, unicode_literals
# This will make sure celery is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app
__all__ = ['celery_app']

Creating your reports modules

We will be creating a reporting module which will be customizable by the customer through the Django admin interface.

Defining the models for the scheduled reports

We need the administrator of the app to customize the reports scheduling using cron expressions, to evaluate the cron expressions we are using croniter.

class ScheduledReport(models.Model):
    """
        Contains email subject and cron expression,to evaluate when the email has to be sent
    """
    subject = models.CharField(max_length=200)
    last_run_at = models.DateTimeField(null=True, blank=True)
    next_run_at = models.DateTimeField(null=True, blank=True)
    cron_expression = models.CharField(max_length=200)
    def save(self, *args, **kwargs):
        """
        function to evaluate "next_run_at" using the cron expression, so that it is updated once the report is sent.
        """
        self.last_run_at = datetime.now()
        iter = croniter(self.cron_expression, self.last_run_at)
        self.next_run_at = iter.get_next(datetime)
        super(ScheduledReport, self).save(*args, **kwargs)
    def __unicode__(self):
        return self.subject
class ScheduledReportGroup(models.Model):
    """
        Many to many mapping between reports which will be sent out in a
        scheduled report
    """
    report = models.ForeignKey(Report, related_name='report')
    scheduled_report = models.ForeignKey(ScheduledReport,
                               related_name='relatedscheduledreport')
class ReportRecipient(models.Model):
    """
        Stores all the recipients of the given scheduled report
    """
    email = models.EmailField()
    scheduled_report = models.ForeignKey(ScheduledReport, related_name='reportrecep')

The administrator can schedule the reports from the admin interface by entering a cron expression, based on which the reports will be sent out.

Overriding Django form for validation

We will be creating a custom form so that we can validate that the cron expression entered by the user is valid. We are going to hide the last_run_at and next_run_at from the user as it would be irrelevant to them.

from datetime import datetime
from croniter import croniter
from django.forms import ModelForm, ValidationError
from models import ScheduledReport
class ScheduledReportForm(ModelForm):
    class Meta:
        model = ScheduledReport
        fields = ['subject', 'cron_expression']
        fields = ['subject', 'cron_expression']
        help_texts = {'cron_expression': 'Scheduled time is considered in UTC'}
    def clean(self):
        cleaned_data = super(ScheduledReportForm, self).clean()
        cron_expression = cleaned_data.get("cron_expression")
        try:
            iter = croniter(cron_expression, datetime.now())
        except:
            raise ValidationError("Incorrect cron expression:\
            The information you must include is (in order of appearance):\
            A number (or list of numbers, or range of numbers), m, representing the minute of the hour\
            A number (or list of numbers, or range of numbers), h, representing the hour of the day\
            A number (or list of numbers, or range of numbers), dom, representing the day of the month\
            A number (or list, or range), or name (or list of names), mon, representing the month of the year\
            A number (or list, or range), or name (or list of names), dow, representing the day of the week\
            The asterisks (*) in our entry tell cron that for that unit of time, the job should be run every.\
            Eg. */5 * * * * cron for executing every 5 mins")
        return cleaned_data

Creating an admin interface for the Scheduled Reports

from django.contrib import admin
from django.template.defaultfilters import escape
from django.core.urlresolvers import reverse
from project.models import ScheduledReport, ReportRecipient, ScheduledReportGroup
from forms import ScheduledReportForm
class ReportRecipientAdmin(admin.TabularInline):
    model = ReportRecipient
class ScheduledReportAdmin(admin.ModelAdmin):
    """
        List display for Scheduled reports in Django admin
    """
    model = ScheduledReport
    list_display = ('id', 'get_recipients')
    inlines = [
        ReportRecipientAdmin
    ]
    form = ScheduledReportForm
    def get_recipients(self, model):
        recipients = model.reportrecep.all().values_list('email', flat=True)
        if not recipients:
            return 'No recipients added'
        recipient_list = ''
        for recipient in recipients:
            recipient_list = recipient_list + recipient + ', '
        return recipient_list[:-2]
    get_recipients.short_description = 'Recipients'
    get_recipients.allow_tags = True
class ScheduledReportGroupAdmin(admin.ModelAdmin):
    """
        List display for ScheduledReportGroup Admin
    """
    model = ScheduledReportGroup
    list_display = ('get_scheduled_report_name','get_report_name')
    def get_scheduled_report_name(self, model):
        return model.scheduled_report.subject
    def get_report_name(self, model):
        return model.report.name
    get_scheduled_report_name.short_description = "Scheduled Report Name"
    get_report_name.short_description = "Report Name"
    show_change_link = True
    get_report_name.allow_tags = True
admin.site.register(ScheduledReport, ScheduledReportAdmin)
admin.site.register(ScheduledReportGroup, ScheduledReportGroupAdmin)

Creating your reports email service

Create a file project/project/email_service.py. This module consists of the scheduled reports’ emailing service.

from datetime import datetime, timedelta
from django.core.mail import send_mail
from django.template import Template, Context
from django.http import HttpResponse
from django.conf import settings
from .models import ScheduledReport, ScheduledReportGroup, ReportRecipient
class ScheduledReportConfig(object):
    def __init__(self, scheduled_report):
        """
            Expects a scheduled report object and inititializes
            its own scheduled_report attribute with it
        """
        self.scheduled_report = scheduled_report
    def get_report_config(self):
        """
            Returns the configuration related to a scheduled report, needed
            to populate the email
        """
        return {
                "template_context": self._get_related_reports_data(),
                "recipients": self._get_report_recipients()
                }
    def _get_related_reports_data(self):
        """
            Returns the list of reports data which needs to be sent out in a scheduled report
        """
        //Logic to get the reports data and format it as you need
        pass
    def _get_report_recipients(self):
        """
            Returns the recipient list for a scheduled report
        """
        //Logic to get the recipients for a scheduled report
        pass
def create_email_data(content=None):
    //Generate html for the the email body
    content = '''
            
        
         ''' + str(content) + ''''''
    return content
def send_emails():
        current_time = datetime.utcnow()
        //Get all the reports which have to sent out till the current time.
        scheduled_reports = ScheduledReport.objects.filter(next_run_at__lt = current_time)
        for scheduled_report in scheduled_reports:
            report_config = ScheduledReportConfig(scheduled_report).get_report_config()
            //Specify the template path you want to send out in the email.
            template = Template(create_email_data('path/to/your/email_template.html'))
            //Create your email html using Django's context processor
            report_template = template.render(Context(report_config['template_context']))
            scheduled_report.save()
            if not scheduled_report.subject:
                //Handle exception for subject not provided
            if not report_config['recipients']:
                //Handle exception for recipients not provided
            send_mail(
                scheduled_report.subject, 'Here is the message.',
                settings.EMAIL_HOST_USER, report_config['recipients'],
                fail_silently=False, html_message=report_template
            )

Scheduling your email service

Once our email service is ready, we need to schedule the email service in celery. Create a tasks.py file in project/project. We are using celery’s cron based periodic tasks for scheduling our reports.

from celery.task.schedules import crontab
from celery.decorators import periodic_task
from email_service import send_emails
# this will run every minute, see http://celeryproject.org/docs/reference/celery.task.schedules.html#celery.task.schedules.crontab
@periodic_task(run_every=crontab(hour="*", minute="*", day_of_week="*"))
def trigger_emails():
    send_emails()

We have scheduled to run the email service every minute, you can change it based on your requirements. The celery worker will call the email service every minute and whichever reports are due, will be sent out and, their next_run_at and last_run_at attribute will be updated based on the cron expression.

Getting your service worker up and running

Running your celery worker is as simple as running a Django server, just run the command: python manage.py celery worker –beat –loglevel=info –without-gossip –without-mingle –without-heartbeat Your worker is up and running in the background. Note: We are using Django’s broker only for development for production, we would need Redis, RabbitMQ or some other broker service which is robust and scalable. Also in production, you would need to run celery as a daemon. Here is the link to the documentation for Daemonization.

Summary

We now have a scheduled reporting module which provides the application’s administrator enough flexibility to schedule reports through Django’s admin interface, by simply specifying the cron expression as to when the report needs to be sent out.